Add _source-only snapshot repository (#32844)

docs/reference/modules/snapshots.asciidoc

@@ -207,6 +207,51 @@ repositories.url.allowed_urls: ["http://www.example.org/root/*", "https://*.mydo
 URL repositories with `file:` URLs can only point to locations registered in the `path.repo` setting similar to
 shared file system repository.
 
+[float]
+[role="xpack"]
+[testenv="basic"]
+===== Source Only Repository
+
+A source repository enables you to create minimal, source-only snapshots that take up to 50% less space on disk.
+Source only snapshots contain stored fields and index metadata. They do not include index or doc values structures
+and are not searchable when restored. After restoring a source-only snapshot, you must <<docs-reindex,reindex>>
+the data into a new index.
+
+Source repositories delegate to another snapshot repository for storage.
+
+
+[IMPORTANT]
+==================================================
+
+Source only snapshots are only supported if the `_source` field is enabled and no source-filtering is applied.
+When you restore a source only snapshot:
+
+ * The restored index is read-only and can only serve `match_all` search or scroll requests to enable reindexing.
+
+ * Queries other than `match_all` and `_get` requests are not supported.
+
+ * The mapping of the restored index is empty, but the original mapping is available from the types top
+   level `meta` element.
+
+==================================================
+
+When you create a source repository, you must specify the type and name of the delegate repository
+where the snapshots will be stored:
+
+[source,js]
+-----------------------------------
+PUT _snapshot/my_src_only_repository
+{
+  "type": "source",
+  "settings": {
+    "delegate_type": "fs",
+    "location": "my_backup_location"
+  }
+}
+-----------------------------------
+// CONSOLE
+// TEST[continued]
+
 [float]
 ===== Repository plugins
 

libs/core/src/main/java/org/elasticsearch/core/internal/io/IOUtils.java

@@ -20,6 +20,7 @@
 import java.io.Closeable;
 import java.io.IOException;
 import java.nio.channels.FileChannel;
+import java.nio.charset.StandardCharsets;
 import java.nio.file.FileVisitResult;
 import java.nio.file.FileVisitor;
 import java.nio.file.Files;
@@ -36,6 +37,14 @@
  */
 public final class IOUtils {
 
+    /**
+     * UTF-8 charset string.
+     * <p>Where possible, use {@link StandardCharsets#UTF_8} instead,
+     * as using the String constant may slow things down.
+     * @see StandardCharsets#UTF_8
+     */
+    public static final String UTF_8 = StandardCharsets.UTF_8.name();
+
     private IOUtils() {
         // Static utils methods
     }

server/src/main/java/org/elasticsearch/index/engine/Engine.java

@@ -1600,7 +1600,7 @@ public static class IndexCommitRef implements Closeable {
         private final CheckedRunnable<IOException> onClose;
         private final IndexCommit indexCommit;
 
-        IndexCommitRef(IndexCommit indexCommit, CheckedRunnable<IOException> onClose) {
+        public IndexCommitRef(IndexCommit indexCommit, CheckedRunnable<IOException> onClose) {
             this.indexCommit = indexCommit;
             this.onClose = onClose;
         }

server/src/main/java/org/elasticsearch/index/engine/EngineFactory.java

@@ -21,6 +21,7 @@
 /**
  * Simple Engine Factory
  */
+@FunctionalInterface
 public interface EngineFactory {
 
     Engine newReadWriteEngine(EngineConfig config);

server/src/main/java/org/elasticsearch/index/seqno/SeqNoStats.java

@@ -91,5 +91,4 @@ public String toString() {
             ", globalCheckpoint=" + globalCheckpoint +
             '}';
     }
-
 }

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

@@ -51,6 +51,4 @@ public IndexSettings indexSettings() {
     public String nodeName() {
         return indexSettings.getNodeName();
     }
-
-
 }

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

@@ -1438,14 +1438,31 @@ public void createEmpty() throws IOException {
      * This is used to make sure no existing shard will recovery from this index using ops based recovery.
      */
     public void bootstrapNewHistory() throws IOException {
+        metadataLock.writeLock().lock();
+        try {
+            Map<String, String> userData = readLastCommittedSegmentsInfo().getUserData();
+            final SequenceNumbers.CommitInfo seqno = SequenceNumbers.loadSeqNoInfoFromLuceneCommit(userData.entrySet());
+            bootstrapNewHistory(seqno.maxSeqNo);
+        } finally {
+            metadataLock.writeLock().unlock();
+        }
+    }
+
+    /**
+     * Marks an existing lucene index with a new history uuid and sets the given maxSeqNo as the local checkpoint
+     * as well as the maximum sequence number.
+     * This is used to make sure no existing shard will recovery from this index using ops based recovery.
+     * @see SequenceNumbers#LOCAL_CHECKPOINT_KEY
+     * @see SequenceNumbers#MAX_SEQ_NO
+     */
+    public void bootstrapNewHistory(long maxSeqNo) throws IOException {
         metadataLock.writeLock().lock();
         try (IndexWriter writer = newIndexWriter(IndexWriterConfig.OpenMode.APPEND, directory, null)) {
             final Map<String, String> userData = getUserData(writer);
-            final SequenceNumbers.CommitInfo seqno = SequenceNumbers.loadSeqNoInfoFromLuceneCommit(userData.entrySet());
             final Map<String, String> map = new HashMap<>();
             map.put(Engine.HISTORY_UUID_KEY, UUIDs.randomBase64UUID());
-            map.put(SequenceNumbers.MAX_SEQ_NO, Long.toString(seqno.maxSeqNo));
-            map.put(SequenceNumbers.LOCAL_CHECKPOINT_KEY, Long.toString(seqno.maxSeqNo));
+            map.put(SequenceNumbers.MAX_SEQ_NO, Long.toString(maxSeqNo));
+            map.put(SequenceNumbers.LOCAL_CHECKPOINT_KEY, Long.toString(maxSeqNo));
             logger.debug("bootstrap a new history_uuid [{}], user_data [{}]", map, userData);
             updateCommitData(writer, map);
         } finally {

server/src/main/java/org/elasticsearch/indices/IndicesService.java

@@ -396,7 +396,6 @@ public boolean hasIndex(Index index) {
     public IndexService indexService(Index index) {
         return indices.get(index.getUUID());
     }
-
     /**
      * Returns an IndexService for the specified index if exists otherwise a {@link IndexNotFoundException} is thrown.
      */

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

@@ -0,0 +1,167 @@
+/*
+ * Licensed to Elasticsearch under one or more contributor
+ * license agreements. See the NOTICE file distributed with
+ * this work for additional information regarding copyright
+ * ownership. Elasticsearch licenses this file to you under
+ * the Apache License, Version 2.0 (the "License"); you may
+ * not use this file except in compliance with the License.
+ * You may obtain a copy of the License at
+ *
+ *    http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing,
+ * software distributed under the License is distributed on an
+ * "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
+ * KIND, either express or implied.  See the License for the
+ * specific language governing permissions and limitations
+ * under the License.
+ */
+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 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.