Introduce repository test kit/analyser

docs/reference/snapshot-restore/apis/snapshot-restore-apis.asciidoc

@@ -27,6 +27,7 @@ For more information, see <<snapshot-restore>>.
 
 include::put-repo-api.asciidoc[]
 include::verify-repo-api.asciidoc[]
+include::repo-analysis-api.asciidoc[]
 include::get-repo-api.asciidoc[]
 include::delete-repo-api.asciidoc[]
 include::clean-up-repo-api.asciidoc[]

docs/reference/snapshot-restore/register-repository.asciidoc

@@ -260,6 +260,9 @@ POST /_snapshot/my_unverified_backup/_verify
 
 It returns a list of nodes where repository was successfully verified or an error message if verification process failed.
 
+If desired, you can also test a repository more thoroughly using the
+<<repo-analysis-api,Repository analysis API>>.
+
 [discrete]
 [[snapshots-repository-cleanup]]
 === Repository cleanup

server/src/main/java/org/elasticsearch/common/blobstore/fs/FsBlobContainer.java

@@ -37,6 +37,7 @@
 import java.nio.file.attribute.BasicFileAttributes;
 import java.util.Collections;
 import java.util.HashMap;
+import java.util.Iterator;
 import java.util.List;
 import java.util.Map;
 import java.util.concurrent.atomic.AtomicLong;
@@ -88,7 +89,7 @@ public Map<String, BlobMetadata> listBlobsByPrefix(String blobNamePrefix) throws
         Map<String, BlobMetadata> builder = new HashMap<>();
 
         blobNamePrefix = blobNamePrefix == null ? "" : blobNamePrefix;
-        try (DirectoryStream<Path> stream = Files.newDirectoryStream(path, blobNamePrefix + "*")) {
+        try (DirectoryStream<Path> stream = newDirectoryStreamIfFound(blobNamePrefix)) {
             for (Path file : stream) {
                 final BasicFileAttributes attrs;
                 try {
@@ -105,6 +106,34 @@ public Map<String, BlobMetadata> listBlobsByPrefix(String blobNamePrefix) throws
         return unmodifiableMap(builder);
     }
 
+    private DirectoryStream<Path> newDirectoryStreamIfFound(String blobNamePrefix) throws IOException {
+        try {
+            return Files.newDirectoryStream(path, blobNamePrefix + "*");
+        } catch (FileNotFoundException | NoSuchFileException e) {
+            // a nonexistent directory contains no blobs
+            return new DirectoryStream<>() {
+                @Override
+                public Iterator<Path> iterator() {
+                    return new Iterator<>() {
+                        @Override
+                        public boolean hasNext() {
+                            return false;
+                        }
+
+                        @Override
+                        public Path next() {
+                            return null;
+                        }
+                    };
+                }
+
+                @Override
+                public void close() {
+                }
+            };
+        }
+    }
+
     @Override
     public DeleteResult delete() throws IOException {
         final AtomicLong filesDeleted = new AtomicLong(0L);

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

@@ -380,7 +380,7 @@ protected void doRun() {
         });
     }
 
-    static boolean isDedicatedVotingOnlyNode(Set<DiscoveryNodeRole> roles) {
+    public static boolean isDedicatedVotingOnlyNode(Set<DiscoveryNodeRole> roles) {
         return roles.contains(DiscoveryNodeRole.MASTER_ROLE) && roles.contains(DiscoveryNodeRole.DATA_ROLE) == false &&
             roles.stream().anyMatch(role -> role.roleName().equals("voting_only"));
     }

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

@@ -35,5 +35,11 @@ public RestStatus status() {
     public RepositoryVerificationException(StreamInput in) throws IOException{
         super(in);
     }
+
+    @Override
+    public synchronized Throwable fillInStackTrace() {
+        // stack trace for a verification failure is uninteresting, the message has all the information we need
+        return this;
+    }
 }
 

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

@@ -2471,17 +2471,47 @@ private static ActionListener<Void> fileQueueListener(BlockingQueue<BlobStoreInd
         });
     }
 
-    private static InputStream maybeRateLimit(InputStream stream, Supplier<RateLimiter> rateLimiterSupplier, CounterMetric metric) {
-        return new RateLimitingInputStream(stream, rateLimiterSupplier, metric::inc);
+    private static InputStream maybeRateLimit(
+            InputStream stream,
+            Supplier<RateLimiter> rateLimiterSupplier,
+            RateLimitingInputStream.Listener throttleListener) {
+        return new RateLimitingInputStream(stream, rateLimiterSupplier, throttleListener);
     }
 
+    /**
+     * Wrap the restore rate limiter (controlled by the repository setting `max_restore_bytes_per_sec` and the cluster setting
+     * `indices.recovery.max_bytes_per_sec`) around the given stream. Any throttling is reported to the given listener and not otherwise
+     * recorded in the value returned by {@link BlobStoreRepository#getRestoreThrottleTimeInNanos}.
+     */
     public InputStream maybeRateLimitRestores(InputStream stream) {
-        return maybeRateLimit(maybeRateLimit(stream, () -> restoreRateLimiter, restoreRateLimitingTimeInNanos),
-            recoverySettings::rateLimiter, restoreRateLimitingTimeInNanos);
+        return maybeRateLimitRestores(stream, restoreRateLimitingTimeInNanos::inc);
     }
 
+    /**
+     * Wrap the restore rate limiter (controlled by the repository setting `max_restore_bytes_per_sec` and the cluster setting
+     * `indices.recovery.max_bytes_per_sec`) around the given stream. Any throttling is recorded in the value returned by {@link
+     * BlobStoreRepository#getRestoreThrottleTimeInNanos}.
+     */
+    public InputStream maybeRateLimitRestores(InputStream stream, RateLimitingInputStream.Listener throttleListener) {
+        return maybeRateLimit(maybeRateLimit(stream, () -> restoreRateLimiter, throttleListener),
+            recoverySettings::rateLimiter, throttleListener);
+    }
+
+    /**
+     * Wrap the snapshot rate limiter (controlled by the repository setting `max_snapshot_bytes_per_sec`) around the given stream. Any
+     * throttling is recorded in the value returned by {@link BlobStoreRepository#getSnapshotThrottleTimeInNanos()}.
+     */
     public InputStream maybeRateLimitSnapshots(InputStream stream) {
-        return maybeRateLimit(stream, () -> snapshotRateLimiter, snapshotRateLimitingTimeInNanos);
+        return maybeRateLimitSnapshots(stream, snapshotRateLimitingTimeInNanos::inc);
+    }
+
+    /**
+     * Wrap the snapshot rate limiter (controlled by the repository setting `max_snapshot_bytes_per_sec`) around the given stream. Any
+     * throttling is reported to the given listener and not otherwise recorded in the value returned by {@link
+     * BlobStoreRepository#getSnapshotThrottleTimeInNanos()}.
+     */
+    public InputStream maybeRateLimitSnapshots(InputStream stream, RateLimitingInputStream.Listener throttleListener) {
+        return maybeRateLimit(stream, () -> snapshotRateLimiter, throttleListener);
     }
 
     @Override
@@ -2721,6 +2751,10 @@ private static void failStoreIfCorrupted(Store store, Exception e) {
         }
     }
 
+    public boolean supportURLRepo() {
+        return supportURLRepo;
+    }
+
     /**
      * The result of removing a snapshot from a shard folder in the repository.
      */

server/src/main/java/org/elasticsearch/rest/RestRequest.java

@@ -360,6 +360,18 @@ public float paramAsFloat(String key, float defaultValue) {
         }
     }
 
+    public double paramAsDouble(String key, double defaultValue) {
+        String sValue = param(key);
+        if (sValue == null) {
+            return defaultValue;
+        }
+        try {
+            return Double.parseDouble(sValue);
+        } catch (NumberFormatException e) {
+            throw new IllegalArgumentException("Failed to parse double parameter [" + key + "] with value [" + sValue + "]", e);
+        }
+    }
+
     public int paramAsInt(String key, int defaultValue) {
         String sValue = param(key);
         if (sValue == null) {

x-pack/plugin/security/src/main/java/org/elasticsearch/xpack/security/operator/OperatorOnlyRegistry.java

@@ -33,7 +33,12 @@ public class OperatorOnlyRegistry {
         "cluster:admin/autoscaling/put_autoscaling_policy",
         "cluster:admin/autoscaling/delete_autoscaling_policy",
         "cluster:admin/autoscaling/get_autoscaling_policy",
-        "cluster:admin/autoscaling/get_autoscaling_capacity");
+        "cluster:admin/autoscaling/get_autoscaling_capacity",
+        // Repository analysis actions are not mentioned in core, literal strings are needed.
+        "cluster:admin/repository/analyze",
+        "cluster:admin/repository/analyze/blob",
+        "cluster:admin/repository/analyze/blob/read"
+        );
 
     private final ClusterSettings clusterSettings;
 

x-pack/plugin/snapshot-repo-test-kit/build.gradle

@@ -0,0 +1,30 @@
+apply plugin: 'elasticsearch.internal-cluster-test'
+apply plugin: 'elasticsearch.esplugin'
+esplugin {
+  name 'snapshot-repo-test-kit'
+  description 'A plugin for a test kit for snapshot repositories'
+  classname 'org.elasticsearch.repositories.blobstore.testkit.SnapshotRepositoryTestKit'
+  extendedPlugins = ['x-pack-core']
+}
+archivesBaseName = 'x-pack-snapshot-repo-test-kit'
+
+dependencies {
+  compileOnly project(path: xpackModule('core'), configuration: 'default')
+  internalClusterTestImplementation project(path: xpackModule('core'), configuration: 'testArtifacts')
+}
+
+addQaCheckDependencies()
+
+configurations {
+  testArtifacts.extendsFrom testRuntime
+  testArtifacts.extendsFrom testImplementation
+}
+
+def testJar = tasks.register("testJar", Jar) {
+  appendix 'test'
+  from sourceSets.test.output
+}
+
+artifacts {
+  testArtifacts testJar
+}

x-pack/plugin/snapshot-repo-test-kit/qa/build.gradle

@@ -0,0 +1,6 @@
+apply plugin: 'elasticsearch.build'
+tasks.named("test").configure { enabled = false }
+
+dependencies {
+  api project(':test:framework')
+}

x-pack/plugin/snapshot-repo-test-kit/qa/rest/build.gradle

@@ -0,0 +1,27 @@
+apply plugin: 'elasticsearch.testclusters'
+apply plugin: 'elasticsearch.standalone-rest-test'
+apply plugin: 'elasticsearch.rest-test'
+apply plugin: 'elasticsearch.rest-resources'
+
+dependencies {
+  testImplementation project(path: xpackModule('snapshot-repo-test-kit'), configuration: 'testArtifacts')
+}
+
+// TODO we want 3rd-party tests for other repository types too
+final File repoDir = file("$buildDir/testclusters/repo")
+
+tasks.named("integTest").configure {
+  systemProperty 'tests.path.repo', repoDir
+}
+
+testClusters.matching { it.name == "integTest" }.configureEach {
+  testDistribution = 'DEFAULT'
+  setting 'path.repo', repoDir.absolutePath
+}
+
+restResources {
+  restApi {
+    includeCore 'indices', 'search', 'bulk', 'snapshot', 'nodes', '_common'
+    includeXpack 'snapshot_repo_test_kit'
+  }
+}

x-pack/plugin/snapshot-repo-test-kit/qa/rest/src/test/java/org/elasticsearch/repositories/blobstore/testkit/rest/FsSnapshotRepoTestKitIT.java

@@ -0,0 +1,25 @@
+/*
+ * Copyright Elasticsearch B.V. and/or licensed to Elasticsearch B.V. under one
+ * or more contributor license agreements. Licensed under the Elastic License
+ * 2.0; you may not use this file except in compliance with the Elastic License
+ * 2.0.
+ */
+
+package org.elasticsearch.repositories.blobstore.testkit.rest;
+
+import org.elasticsearch.common.settings.Settings;
+import org.elasticsearch.repositories.blobstore.testkit.AbstractSnapshotRepoTestKitRestTestCase;
+import org.elasticsearch.repositories.fs.FsRepository;
+
+public class FsSnapshotRepoTestKitIT extends AbstractSnapshotRepoTestKitRestTestCase {
+
+    @Override
+    protected String repositoryType() {
+        return FsRepository.TYPE;
+    }
+
+    @Override
+    protected Settings repositorySettings() {
+        return Settings.builder().put("location", System.getProperty("tests.path.repo")).build();
+    }
+}

x-pack/plugin/snapshot-repo-test-kit/qa/rest/src/test/java/org/elasticsearch/repositories/blobstore/testkit/rest/SnapshotRepoTestKitClientYamlTestSuiteIT.java

@@ -0,0 +1,24 @@
+/*
+ * Copyright Elasticsearch B.V. and/or licensed to Elasticsearch B.V. under one
+ * or more contributor license agreements. Licensed under the Elastic License
+ * 2.0; you may not use this file except in compliance with the Elastic License
+ * 2.0.
+ */
+
+package org.elasticsearch.repositories.blobstore.testkit.rest;
+
+import com.carrotsearch.randomizedtesting.annotations.ParametersFactory;
+import org.elasticsearch.test.rest.yaml.ClientYamlTestCandidate;
+import org.elasticsearch.test.rest.yaml.ESClientYamlSuiteTestCase;
+
+public class SnapshotRepoTestKitClientYamlTestSuiteIT extends ESClientYamlSuiteTestCase {
+
+    public SnapshotRepoTestKitClientYamlTestSuiteIT(final ClientYamlTestCandidate testCandidate) {
+        super(testCandidate);
+    }
+
+    @ParametersFactory
+    public static Iterable<Object[]> parameters() throws Exception {
+        return ESClientYamlSuiteTestCase.createParameters();
+    }
+}

x-pack/plugin/snapshot-repo-test-kit/qa/rest/src/test/resources/rest-api-spec/api/snapshot.repository_analyze.json

@@ -0,0 +1,68 @@
+{
+  "snapshot.repository_analyze":{
+    "documentation":{
+      "url":"https://www.elastic.co/guide/en/elasticsearch/reference/master/modules-snapshots.html",
+      "description":"Analyzes a repository for correctness and performance"
+    },
+    "stability":"stable",
+    "visibility":"public",
+    "url":{
+      "paths":[
+        {
+          "path":"/_snapshot/{repository}/_analyze",
+          "methods":[
+            "POST"
+          ],
+          "parts":{
+            "repository":{
+              "type":"string",
+              "description":"A repository name"
+            }
+          }
+        }
+      ]
+    },
+    "params":{
+      "blob_count":{
+        "type":"number",
+        "description":"Number of blobs to create during the test. Defaults to 100."
+      },
+      "concurrency":{
+        "type":"number",
+        "description":"Number of operations to run concurrently during the test. Defaults to 10."
+      },
+      "read_node_count":{
+        "type":"number",
+        "description":"Number of nodes on which to read a blob after writing. Defaults to 10."
+      },
+      "early_read_node_count":{
+        "type":"number",
+        "description":"Number of nodes on which to perform an early read on a blob, i.e. before writing has completed. Early reads are rare actions so the 'rare_action_probability' parameter is also relevant. Defaults to 2."
+      },
+      "seed":{
+        "type":"number",
+        "description":"Seed for the random number generator used to create the test workload. Defaults to a random value."
+      },
+      "rare_action_probability":{
+        "type":"number",
+        "description":"Probability of taking a rare action such as an early read or an overwrite. Defaults to 0.02."
+      },
+      "max_blob_size":{
+        "type":"string",
+        "description":"Maximum size of a blob to create during the test, e.g '1gb' or '100mb'. Defaults to '10mb'."
+      },
+      "max_total_data_size":{
+        "type":"string",
+        "description":"Maximum total size of all blobs to create during the test, e.g '1tb' or '100gb'. Defaults to '1gb'."
+      },
+      "timeout":{
+        "type":"time",
+        "description":"Explicit operation timeout. Defaults to '30s'."
+      },
+      "detailed":{
+        "type":"boolean",
+        "description":"Whether to return detailed results or a summary. Defaults to 'false' so that only the summary is returned."
+      }
+    }
+  }
+}