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();