Add _source-only snapshot repository

docs/plugins/repository-source-only.asciidoc

@@ -0,0 +1,39 @@
+[[repository-src-only]]
+=== Source Only Repository
+
+The Source Only repository adds support for creating `_source` only snapshots using any other
+available repository as it's storage backend. This allows using {ref}/modules-snapshots.html[Snapshot/Restore]
+to create incremental, storage optimized, and minimal snapshots of an index.
+
+
+[[repository-src-only-usage]]
+==== Configuration
+
+The `_source` only repository always requires a delegate repository to be used as it's storage backend.
+In order to use the `fs` respository:
+
+[source,js]
+-----------------------------------
+PUT _snapshot/my_src-only_repository
+{
+  "type": "source",
+  "settings": {
+    "delegate_type": "fs",
+    "location": "my_backup_location"
+  }
+}
+-----------------------------------
+// CONSOLE
+// TESTSETUP
+
+Since the `_source` only repository doesn't snapshot any index or doc-values structures but only stored
+fields and index metadata, it's required to reindex the data during the restore process. This can either happen
+as a full re-index based on the mapping of the original index or in a minimal form were only the internal data-structures
+are recreated like the `_id` field in order to update the index. The latter can be configured in the repository settings
+by setting `"minimal": true`. This allows updates and get operations but won't allow for aggregations or searches.
+
+A minimal restore is useful if the data is only needed to be re-indexed into another index or if individual documents should be
+modified or deleted.
+
+During restore the re-indexing progress can be monitored via <<indices-recovery,recovery>> API that shows the per-document progress
+under the `translog` recovery phase.

server/src/main/java/org/elasticsearch/index/shard/StoreRecovery.java

@@ -464,6 +464,7 @@ private void restore(final IndexShard indexShard, final Repository repository, f
             assert indexShard.shardRouting.primary() : "only primary shards can recover from store";
             indexShard.openEngineAndRecoverFromTranslog();
             indexShard.getEngine().fillSeqNoGaps(indexShard.getPendingPrimaryTerm());
+            repository.applyPostRestoreOps(indexShard);
             indexShard.finalizeRecovery();
             indexShard.postRecovery("restore done");
         } catch (Exception e) {

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

@@ -155,12 +155,15 @@ protected void closeInternal() {
     public Store(ShardId shardId, IndexSettings indexSettings, DirectoryService directoryService, ShardLock shardLock) throws IOException {
         this(shardId, indexSettings, directoryService, shardLock, OnClose.EMPTY);
     }
-
     public Store(ShardId shardId, IndexSettings indexSettings, DirectoryService directoryService, ShardLock shardLock,
                  OnClose onClose) throws IOException {
+        this(shardId, indexSettings, directoryService.newDirectory(), shardLock, onClose);
+    }
+
+    public Store(ShardId shardId, IndexSettings indexSettings, Directory dir, ShardLock shardLock,
+                 OnClose onClose) throws IOException {
         super(shardId, indexSettings);
         final Settings settings = indexSettings.getSettings();
-        Directory dir = directoryService.newDirectory();
         final TimeValue refreshInterval = indexSettings.getValue(INDEX_STORE_STATS_REFRESH_INTERVAL_SETTING);
         logger.debug("store stats are refreshed with refresh_interval [{}]", refreshInterval);
         ByteSizeCachingDirectory sizeCachingDir = new ByteSizeCachingDirectory(dir, refreshInterval);

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

@@ -0,0 +1,154 @@
+package org.elasticsearch.repositories;
+
+import org.apache.lucene.index.IndexCommit;
+import org.elasticsearch.Version;
+import org.elasticsearch.cluster.metadata.IndexMetaData;
+import org.elasticsearch.cluster.metadata.MetaData;
+import org.elasticsearch.cluster.metadata.RepositoryMetaData;
+import org.elasticsearch.cluster.node.DiscoveryNode;
+import org.elasticsearch.common.component.Lifecycle;
+import org.elasticsearch.common.component.LifecycleListener;
+import org.elasticsearch.index.shard.IndexShard;
+import org.elasticsearch.index.shard.ShardId;
+import org.elasticsearch.index.snapshots.IndexShardSnapshotStatus;
+import org.elasticsearch.index.store.Store;
+import org.elasticsearch.indices.recovery.RecoveryState;
+import org.elasticsearch.snapshots.SnapshotId;
+import org.elasticsearch.snapshots.SnapshotInfo;
+import org.elasticsearch.snapshots.SnapshotShardFailure;
+
+import java.io.IOException;
+import java.util.List;
+
+public class FilterRepository implements Repository {
+
+    private final Repository in;
+
+    public FilterRepository(Repository in) {
+        this.in = in;
+    }
+
+    @Override
+    public RepositoryMetaData getMetadata() {
+        return in.getMetadata();
+    }
+
+    @Override
+    public SnapshotInfo getSnapshotInfo(SnapshotId snapshotId) {
+        return in.getSnapshotInfo(snapshotId);
+    }
+
+    @Override
+    public MetaData getSnapshotGlobalMetaData(SnapshotId snapshotId) {
+        return in.getSnapshotGlobalMetaData(snapshotId);
+    }
+
+    @Override
+    public IndexMetaData getSnapshotIndexMetaData(SnapshotId snapshotId, IndexId index) throws IOException {
+        return in.getSnapshotIndexMetaData(snapshotId, index);
+    }
+
+    @Override
+    public RepositoryData getRepositoryData() {
+        return in.getRepositoryData();
+    }
+
+    @Override
+    public void initializeSnapshot(SnapshotId snapshotId, List<IndexId> indices, MetaData metaData) {
+        in.initializeSnapshot(snapshotId, indices, metaData);
+    }
+
+    @Override
+    public SnapshotInfo finalizeSnapshot(SnapshotId snapshotId, List<IndexId> indices, long startTime, String failure, int totalShards,
+                                         List<SnapshotShardFailure> shardFailures, long repositoryStateId, boolean includeGlobalState) {
+        return in.finalizeSnapshot(snapshotId, indices, startTime, failure, totalShards, shardFailures, repositoryStateId,
+            includeGlobalState);
+    }
+
+    @Override
+    public void deleteSnapshot(SnapshotId snapshotId, long repositoryStateId) {
+        in.deleteSnapshot(snapshotId, repositoryStateId);
+    }
+
+    @Override
+    public long getSnapshotThrottleTimeInNanos() {
+        return in.getSnapshotThrottleTimeInNanos();
+    }
+
+    @Override
+    public long getRestoreThrottleTimeInNanos() {
+        return in.getRestoreThrottleTimeInNanos();
+    }
+
+    @Override
+    public String startVerification() {
+        return in.startVerification();
+    }
+
+    @Override
+    public void endVerification(String verificationToken) {
+        in.endVerification(verificationToken);
+    }
+
+    @Override
+    public void verify(String verificationToken, DiscoveryNode localNode) {
+        in.verify(verificationToken, localNode);
+    }
+
+    @Override
+    public boolean isReadOnly() {
+        return in.isReadOnly();
+    }
+
+    @Override
+    public void snapshotShard(IndexShard shard, Store store, SnapshotId snapshotId, IndexId indexId, IndexCommit snapshotIndexCommit,
+                              IndexShardSnapshotStatus snapshotStatus) {
+        in.snapshotShard(shard, store, snapshotId, indexId, snapshotIndexCommit, snapshotStatus);
+    }
+
+    @Override
+    public void restoreShard(IndexShard shard, SnapshotId snapshotId, Version version, IndexId indexId, ShardId snapshotShardId,
+                             RecoveryState recoveryState) {
+        in.restoreShard(shard, snapshotId, version, indexId, snapshotShardId, recoveryState);
+    }
+
+    @Override
+    public IndexShardSnapshotStatus getShardSnapshotStatus(SnapshotId snapshotId, Version version, IndexId indexId, ShardId shardId) {
+        return in.getShardSnapshotStatus(snapshotId, version, indexId, shardId);
+    }
+
+    @Override
+    public void applyPostRestoreOps(IndexShard shard) throws IOException {
+        in.applyPostRestoreOps(shard);
+    }
+
+    @Override
+    public Lifecycle.State lifecycleState() {
+        return in.lifecycleState();
+    }
+
+    @Override
+    public void addLifecycleListener(LifecycleListener listener) {
+        in.addLifecycleListener(listener);
+    }
+
+    @Override
+    public void removeLifecycleListener(LifecycleListener listener) {
+        in.removeLifecycleListener(listener);
+    }
+
+    @Override
+    public void start() {
+        in.start();
+    }
+
+    @Override
+    public void stop() {
+        in.stop();
+    }
+
+    @Override
+    public void close() {
+        in.close();
+    }
+}

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

@@ -398,7 +398,7 @@ private Repository createRepository(RepositoryMetaData repositoryMetaData) {
                 "repository type [" + repositoryMetaData.type() + "] does not exist");
         }
         try {
-            Repository repository = factory.create(repositoryMetaData);
+            Repository repository = factory.create(repositoryMetaData, typesRegistry::get);
             repository.start();
             return repository;
         } catch (Exception e) {

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

@@ -28,13 +28,15 @@
 import org.elasticsearch.index.shard.IndexShard;
 import org.elasticsearch.index.shard.ShardId;
 import org.elasticsearch.index.snapshots.IndexShardSnapshotStatus;
+import org.elasticsearch.index.store.Store;
 import org.elasticsearch.indices.recovery.RecoveryState;
 import org.elasticsearch.snapshots.SnapshotId;
 import org.elasticsearch.snapshots.SnapshotInfo;
 import org.elasticsearch.snapshots.SnapshotShardFailure;
 
 import java.io.IOException;
 import java.util.List;
+import java.util.function.Function;
 
 /**
  * An interface for interacting with a repository in snapshot and restore.
@@ -46,7 +48,7 @@
  * <ul>
  * <li>Master calls {@link #initializeSnapshot(SnapshotId, List, org.elasticsearch.cluster.metadata.MetaData)}
  * with list of indices that will be included into the snapshot</li>
- * <li>Data nodes call {@link Repository#snapshotShard(IndexShard, SnapshotId, IndexId, IndexCommit, IndexShardSnapshotStatus)}
+ * <li>Data nodes call {@link Repository#snapshotShard(IndexShard, Store, SnapshotId, IndexId, IndexCommit, IndexShardSnapshotStatus)}
  * for each shard</li>
  * <li>When all shard calls return master calls {@link #finalizeSnapshot} with possible list of failures</li>
  * </ul>
@@ -63,6 +65,10 @@ interface Factory {
          * @param metadata    metadata for the repository including name and settings
          */
         Repository create(RepositoryMetaData metadata) throws Exception;
+
+        default Repository create(RepositoryMetaData metaData, Function<String, Repository.Factory> typeLookup) throws Exception {
+            return create(metaData);
+        }
     }
 
     /**
@@ -188,14 +194,15 @@ SnapshotInfo finalizeSnapshot(SnapshotId snapshotId, List<IndexId> indices, long
      * <p>
      * As snapshot process progresses, implementation of this method should update {@link IndexShardSnapshotStatus} object and check
      * {@link IndexShardSnapshotStatus#isAborted()} to see if the snapshot process should be aborted.
-     *
      * @param shard               shard to be snapshotted
+     * @param store               store to be snapshotted
      * @param snapshotId          snapshot id
      * @param indexId             id for the index being snapshotted
      * @param snapshotIndexCommit commit point
      * @param snapshotStatus      snapshot status
      */
-    void snapshotShard(IndexShard shard, SnapshotId snapshotId, IndexId indexId, IndexCommit snapshotIndexCommit, IndexShardSnapshotStatus snapshotStatus);
+    void snapshotShard(IndexShard shard, Store store, SnapshotId snapshotId, IndexId indexId, IndexCommit snapshotIndexCommit,
+                       IndexShardSnapshotStatus snapshotStatus);
 
     /**
      * Restores snapshot of the shard.
@@ -211,6 +218,11 @@ SnapshotInfo finalizeSnapshot(SnapshotId snapshotId, List<IndexId> indices, long
      */
     void restoreShard(IndexShard shard, SnapshotId snapshotId, Version version, IndexId indexId, ShardId snapshotShardId, RecoveryState recoveryState);
 
+    /**
+     * This allows a repository to apply operations after the snapshot has been restored as part of the translog recovery phase.
+     */
+    default void applyPostRestoreOps(IndexShard shard) throws IOException {}
+
     /**
      * Retrieve shard snapshot status for the stored snapshot
      *

server/src/main/java/org/elasticsearch/repositories/blobstore/BlobStoreRepository.java

@@ -846,16 +846,17 @@ private void writeAtomic(final String blobName, final BytesReference bytesRef, b
     }
 
     @Override
-    public void snapshotShard(IndexShard shard, SnapshotId snapshotId, IndexId indexId, IndexCommit snapshotIndexCommit, IndexShardSnapshotStatus snapshotStatus) {
-        SnapshotContext snapshotContext = new SnapshotContext(shard, snapshotId, indexId, snapshotStatus, System.currentTimeMillis());
+    public void snapshotShard(IndexShard shard, Store store, SnapshotId snapshotId, IndexId indexId, IndexCommit snapshotIndexCommit,
+                              IndexShardSnapshotStatus snapshotStatus) {
+        SnapshotContext snapshotContext = new SnapshotContext(store, snapshotId, indexId, snapshotStatus, System.currentTimeMillis());
         try {
             snapshotContext.snapshot(snapshotIndexCommit);
         } catch (Exception e) {
             snapshotStatus.moveToFailed(System.currentTimeMillis(), ExceptionsHelper.detailedMessage(e));
             if (e instanceof IndexShardSnapshotFailedException) {
                 throw (IndexShardSnapshotFailedException) e;
             } else {
-                throw new IndexShardSnapshotFailedException(shard.shardId(), e);
+                throw new IndexShardSnapshotFailedException(store.shardId(), e);
             }
         }
     }
@@ -1158,15 +1159,15 @@ private class SnapshotContext extends Context {
         /**
          * Constructs new context
          *
-         * @param shard          shard to be snapshotted
+         * @param store          store to be snapshotted
          * @param snapshotId     snapshot id
          * @param indexId        the id of the index being snapshotted
          * @param snapshotStatus snapshot status to report progress
          */
-        SnapshotContext(IndexShard shard, SnapshotId snapshotId, IndexId indexId, IndexShardSnapshotStatus snapshotStatus, long startTime) {
-            super(snapshotId, Version.CURRENT, indexId, shard.shardId());
+        SnapshotContext(Store store, SnapshotId snapshotId, IndexId indexId, IndexShardSnapshotStatus snapshotStatus, long startTime) {
+            super(snapshotId, Version.CURRENT, indexId, store.shardId());
             this.snapshotStatus = snapshotStatus;
-            this.store = shard.store();
+            this.store = store;
             this.startTime = startTime;
         }
 

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

@@ -389,7 +389,8 @@ private void snapshot(final IndexShard indexShard, final Snapshot snapshot, fina
         try {
             // we flush first to make sure we get the latest writes snapshotted
             try (Engine.IndexCommitRef snapshotRef = indexShard.acquireLastIndexCommit(true)) {
-                repository.snapshotShard(indexShard, snapshot.getSnapshotId(), indexId, snapshotRef.getIndexCommit(), snapshotStatus);
+                repository.snapshotShard(indexShard, indexShard.store(), snapshot.getSnapshotId(), indexId, snapshotRef.getIndexCommit(),
+                    snapshotStatus);
                 if (logger.isDebugEnabled()) {
                     final IndexShardSnapshotStatus.Copy lastSnapshotStatus = snapshotStatus.asCopy();
                     logger.debug("snapshot ({}) completed to {} with {}", snapshot, repository, lastSnapshotStatus);

server/src/test/java/org/elasticsearch/index/shard/IndexShardTests.java

@@ -2764,7 +2764,8 @@ public boolean isReadOnly() {
         }
 
         @Override
-        public void snapshotShard(IndexShard shard, SnapshotId snapshotId, IndexId indexId, IndexCommit snapshotIndexCommit, IndexShardSnapshotStatus snapshotStatus) {
+        public void snapshotShard(IndexShard shard, Store store, SnapshotId snapshotId, IndexId indexId, IndexCommit snapshotIndexCommit,
+                                  IndexShardSnapshotStatus snapshotStatus) {
         }
 
         @Override

test/framework/src/main/java/org/elasticsearch/index/shard/IndexShardTestCase.java

@@ -131,7 +131,7 @@ public void onRecoveryFailure(RecoveryState state, RecoveryFailedException e, bo
     };
 
     protected ThreadPool threadPool;
-    private long primaryTerm;
+    protected long primaryTerm;
 
     @Override
     public void setUp() throws Exception {
@@ -693,7 +693,8 @@ protected void snapshotShard(final IndexShard shard,
             Index index = shard.shardId().getIndex();
             IndexId indexId = new IndexId(index.getName(), index.getUUID());
 
-            repository.snapshotShard(shard, snapshot.getSnapshotId(), indexId, indexCommitRef.getIndexCommit(), snapshotStatus);
+            repository.snapshotShard(shard, shard.store(), snapshot.getSnapshotId(), indexId, indexCommitRef.getIndexCommit(),
+                snapshotStatus);
         }
 
         final IndexShardSnapshotStatus.Copy lastSnapshotStatus = snapshotStatus.asCopy();