Add repositories metering API

docs/reference/repositories-stats/apis/clear-repositories-stats-archive.asciidoc

@@ -0,0 +1,31 @@
+[[clear-repositories-stats-archive-api]]
+=== Clear repositories statistics archive API
+++++
+<titleabbrev>Clear repositories statistics archive API</titleabbrev>
+++++
+
+Removes the archived repositories statistics in the cluster.
+
+[[clear-repositories-stats-archive-api-request]]
+==== {api-request-title}
+
+`DELETE /_nodes/<node_id>/_repositories_stats`
+
+[[clear-repositories-stats-archive-api-desc]]
+==== {api-description-title}
+
+You can clear the archived repositories utilization statistics using this API.
+
+All the nodes selective options are explained <<cluster-nodes,here>>.
+
+[[clear-repositories-stats-archive-api-path-params]]
+==== {api-path-parms-title}
+
+include::{es-repo-dir}/rest-api/common-parms.asciidoc[tag=node-id]
+
+[[clear-repositories-stats-archive-api-query-params]]
+==== {api-query-parms-title}
+
+[role="child_attributes"]
+[[clear-repositories-stats-archive-api-response-body]]
+==== {api-response-body-title}

docs/reference/repositories-stats/apis/get-repositories-stats.asciidoc

@@ -0,0 +1,142 @@
+[[get-repositories-stats-api]]
+=== Repositories statistics API
+++++
+<titleabbrev>Repositories statistics</titleabbrev>
+++++
+
+Returns cluster repositories utilization statistics.
+
+[[get-repositories-stats-api-request]]
+==== {api-request-title}
+
+`GET /_nodes/<node_id>/_repositories_stats`
+
+[[get-repositories-stats-api-desc]]
+==== {api-description-title}
+
+You can use the cluster repositories statistics API to retrieve repository utilization statistics in a cluster.
+
+
+All the nodes selective options are explained <<cluster-nodes,here>>.
+
+[[get-repositories-stats-api-path-params]]
+==== {api-path-parms-title}
+
+include::{es-repo-dir}/rest-api/common-parms.asciidoc[tag=node-id]
+
+[[get-repositories-stats-api-query-params]]
+==== {api-query-parms-title}
+
+[role="child_attributes"]
+[[get-repositories-stats-api-response-body]]
+==== {api-response-body-title}
+
+`_nodes`::
+(object)
+Contains statistics about the number of nodes selected by the request.
++
+.Properties of `_nodes`
+[%collapsible%open]
+====
+`total`::
+(integer)
+Total number of nodes selected by the request.
+
+`successful`::
+(integer)
+Number of nodes that responded successfully to the request.
+
+`failed`::
+(integer)
+Number of nodes that rejected the request or failed to respond. If this value
+is not `0`, a reason for the rejection or failure is included in the response.
+====
+
+`cluster_name`::
+(string)
+Name of the cluster. Based on the <<cluster.name>> setting.
+
+`nodes`::
+(object)
+Contains repository utilization statistics for the nodes selected by the request.
++
+.Properties of `nodes`
+[%collapsible%open]
+====
+`<node_id>`::
+(array)
+An array of repository utilization statistics for the node.
++
+.Properties of objects in `node_id`
+[%collapsible%open]
+=====
+`repository_name`::
+(string) 
+Repository name.
+
+`repository_type`::
+(string) 
+Repository type.
+
+`repository_location`::
+(string) 
+Represents an unique location within the repository, 
+i.e. a `bucket` + `base_path`.
+
+`repository_ephemeral_id`::
+(string) 
+An identifier that changes every time the repository is updated.
+
+`repository_started_at`::
+(long)
+Time the repository was created or updated. Recorded in milliseconds
+since the https://en.wikipedia.org/wiki/Unix_time[Unix Epoch].
+
+`repository_stopped_at`::
+(long)
+Time the repository was deleted or updated. Recorded in milliseconds
+since the https://en.wikipedia.org/wiki/Unix_time[Unix Epoch].
+
+`request_counts`::
+(object) 
+An object with the number of request performed against the repository
+grouped by request type.
++
+.Properties of `request_counts` for repository type `Azure`
+[%collapsible%open]
+======
+`HEAD`::
+(long) Number of https://docs.microsoft.com/en-us/rest/api/storageservices/get-blob-properties[Get Blob Properties] requests.
+`GET`::
+(long) Number of https://docs.microsoft.com/en-us/rest/api/storageservices/get-blob[Get Blob] requests.
+`PUT`::
+(long) Number of https://docs.microsoft.com/en-us/rest/api/storageservices/put-blob[Put Blob] requests.
+======
++
+.Properties of `request_counts` for repository type `GCP`
+[%collapsible%open]
+======
+`GET`::
+(long) Number of https://cloud.google.com/storage/docs/json_api/v1/objects/get[GET] requests.
+`LIST`::
+(long) Number of https://cloud.google.com/storage/docs/json_api/v1/objects/list[LIST] requests.
+`PUT`::
+(long) Number of https://cloud.google.com/storage/docs/performing-resumable-uploads[PUT] requests.
+`POST`::
+(long) Number of https://cloud.google.com/storage/docs/json_api/v1/objects/insert[POST] requests.
+======
++
+.Properties of `request_counts` for repository type `S3`
+[%collapsible%open]
+======
+`GET`::
+(long) Number of https://docs.aws.amazon.com/AmazonS3/latest/API/API_GetObject.html[GET] requests.
+`LIST`::
+(long) Number of https://docs.aws.amazon.com/AmazonS3/latest/API/API_ListObjects.html[LIST] requests.
+`PUT`::
+(long) Number of https://docs.aws.amazon.com/AmazonS3/latest/API/API_PutObject.html[PUT] requests.
+`POST`::
+(long) Number of https://docs.aws.amazon.com/AmazonS3/latest/dev/mpuoverview.html[MultiPart] requests.
+======
+=====
+====

docs/reference/repositories-stats/apis/repositories-stats.asciidoc

@@ -0,0 +1,12 @@
+[[repositories-stats-apis]]
+== Repositories utilization statistics APIs
+
+experimental[]
+
+You can use the following APIs to retrieve statistics about the repositories utilization.
+
+* <<get-repositories-stats-api,Get repositories utilization statisticss>>
+* <<clear-repositories-stats-archive-api,Clear repositories utilization statistics archive>>
+
+include::get-repositories-stats.asciidoc[]
+include::clear-repositories-stats-archive.asciidoc[]

plugins/repository-azure/src/main/java/org/elasticsearch/repositories/azure/AzureRepository.java

@@ -33,7 +33,7 @@
 import org.elasticsearch.common.unit.ByteSizeValue;
 import org.elasticsearch.common.xcontent.NamedXContentRegistry;
 import org.elasticsearch.indices.recovery.RecoverySettings;
-import org.elasticsearch.repositories.blobstore.BlobStoreRepository;
+import org.elasticsearch.repositories.blobstore.MeteredBlobStoreRepository;
 
 import java.util.Locale;
 import java.util.function.Function;
@@ -52,7 +52,7 @@
  * <dt>{@code compress}</dt><dd>If set to true metadata files will be stored compressed. Defaults to false.</dd>
  * </dl>
  */
-public class AzureRepository extends BlobStoreRepository {
+public class AzureRepository extends MeteredBlobStoreRepository {
     private static final Logger logger = LogManager.getLogger(AzureRepository.class);
 
     public static final String TYPE = "azure";
@@ -82,7 +82,12 @@ public AzureRepository(
         final AzureStorageService storageService,
         final ClusterService clusterService,
         final RecoverySettings recoverySettings) {
-        super(metadata, namedXContentRegistry, clusterService, recoverySettings, buildBasePath(metadata));
+        super(metadata,
+            namedXContentRegistry,
+            clusterService,
+            recoverySettings,
+            buildBasePath(metadata),
+            Repository.CONTAINER_SETTING.get(metadata.settings()));
         this.chunkSize = Repository.CHUNK_SIZE_SETTING.get(metadata.settings());
         this.storageService = storageService;
 

plugins/repository-gcs/src/main/java/org/elasticsearch/repositories/gcs/GoogleCloudStorageRepository.java

@@ -31,15 +31,15 @@
 import org.elasticsearch.common.xcontent.NamedXContentRegistry;
 import org.elasticsearch.indices.recovery.RecoverySettings;
 import org.elasticsearch.repositories.RepositoryException;
-import org.elasticsearch.repositories.blobstore.BlobStoreRepository;
+import org.elasticsearch.repositories.blobstore.MeteredBlobStoreRepository;
 
 import java.util.function.Function;
 
 import static org.elasticsearch.common.settings.Setting.Property;
 import static org.elasticsearch.common.settings.Setting.byteSizeSetting;
 import static org.elasticsearch.common.settings.Setting.simpleString;
 
-class GoogleCloudStorageRepository extends BlobStoreRepository {
+class GoogleCloudStorageRepository extends MeteredBlobStoreRepository {
     private static final Logger logger = LogManager.getLogger(GoogleCloudStorageRepository.class);
 
     // package private for testing
@@ -72,7 +72,7 @@ class GoogleCloudStorageRepository extends BlobStoreRepository {
         final GoogleCloudStorageService storageService,
         final ClusterService clusterService,
         final RecoverySettings recoverySettings) {
-        super(metadata, namedXContentRegistry, clusterService, recoverySettings, buildBasePath(metadata));
+        super(metadata, namedXContentRegistry, clusterService, recoverySettings, buildBasePath(metadata), getSetting(BUCKET, metadata));
         this.storageService = storageService;
 
         this.chunkSize = getSetting(CHUNK_SIZE, metadata);

plugins/repository-s3/src/main/java/org/elasticsearch/repositories/s3/S3Repository.java

@@ -41,7 +41,7 @@
 import org.elasticsearch.repositories.RepositoryData;
 import org.elasticsearch.repositories.RepositoryException;
 import org.elasticsearch.repositories.ShardGenerations;
-import org.elasticsearch.repositories.blobstore.BlobStoreRepository;
+import org.elasticsearch.repositories.blobstore.MeteredBlobStoreRepository;
 import org.elasticsearch.snapshots.SnapshotId;
 import org.elasticsearch.snapshots.SnapshotInfo;
 import org.elasticsearch.snapshots.SnapshotsService;
@@ -66,7 +66,7 @@
  * <dt>{@code compress}</dt><dd>If set to true metadata files will be stored compressed. Defaults to false.</dd>
  * </dl>
  */
-class S3Repository extends BlobStoreRepository {
+class S3Repository extends MeteredBlobStoreRepository {
     private static final Logger logger = LogManager.getLogger(S3Repository.class);
 
     static final String TYPE = "s3";
@@ -194,7 +194,12 @@ class S3Repository extends BlobStoreRepository {
         final S3Service service,
         final ClusterService clusterService,
         final RecoverySettings recoverySettings) {
-        super(metadata, namedXContentRegistry, clusterService, recoverySettings, buildBasePath(metadata));
+        super(metadata,
+            namedXContentRegistry,
+            clusterService,
+            recoverySettings,
+            buildBasePath(metadata),
+            BUCKET_SETTING.get(metadata.settings()));
         this.service = service;
 
         // Parse and validate the user's S3 Storage Class setting

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

@@ -43,9 +43,12 @@
 import org.elasticsearch.common.Strings;
 import org.elasticsearch.common.component.AbstractLifecycleComponent;
 import org.elasticsearch.common.regex.Regex;
+import org.elasticsearch.common.settings.Setting;
 import org.elasticsearch.common.settings.Settings;
+import org.elasticsearch.common.unit.TimeValue;
 import org.elasticsearch.common.util.concurrent.ConcurrentCollections;
 import org.elasticsearch.core.internal.io.IOUtils;
+import org.elasticsearch.repositories.blobstore.MeteredBlobStoreRepository;
 import org.elasticsearch.threadpool.ThreadPool;
 import org.elasticsearch.transport.TransportService;
 
@@ -56,6 +59,8 @@
 import java.util.HashMap;
 import java.util.List;
 import java.util.Map;
+import java.util.stream.Collectors;
+import java.util.stream.Stream;
 
 /**
  * Service responsible for maintaining and providing access to snapshot repositories on nodes.
@@ -64,6 +69,12 @@ public class RepositoriesService extends AbstractLifecycleComponent implements C
 
     private static final Logger logger = LogManager.getLogger(RepositoriesService.class);
 
+    public static final Setting<TimeValue> REPOSITORIES_STATS_ARCHIVE_RETENTION_PERIOD =
+        Setting.positiveTimeSetting("repositories.stats.archive.retention_period", TimeValue.timeValueHours(2), Setting.Property.NodeScope);
+
+    public static final Setting<Integer> REPOSITORIES_STATS_ARCHIVE_MAX_ARCHIVED_STATS =
+        Setting.intSetting("repositories.stats.archive.max_archived_stats", 100, 0, Setting.Property.NodeScope);
+
     private final Map<String, Repository.Factory> typesRegistry;
     private final Map<String, Repository.Factory> internalTypesRegistry;
 
@@ -75,6 +86,7 @@ public class RepositoriesService extends AbstractLifecycleComponent implements C
 
     private final Map<String, Repository> internalRepositories = ConcurrentCollections.newConcurrentMap();
     private volatile Map<String, Repository> repositories = Collections.emptyMap();
+    private final RepositoriesStatsArchive repositoriesStatsArchive;
 
     public RepositoriesService(Settings settings, ClusterService clusterService, TransportService transportService,
                                Map<String, Repository.Factory> typesRegistry, Map<String, Repository.Factory> internalTypesRegistry,
@@ -89,6 +101,10 @@ public RepositoriesService(Settings settings, ClusterService clusterService, Tra
             clusterService.addHighPriorityApplier(this);
         }
         this.verifyAction = new VerifyNodeRepositoryAction(transportService, clusterService, this);
+        this.repositoriesStatsArchive = new RepositoriesStatsArchive(REPOSITORIES_STATS_ARCHIVE_RETENTION_PERIOD.get(settings),
+            REPOSITORIES_STATS_ARCHIVE_MAX_ARCHIVED_STATS.get(settings),
+            threadPool::relativeTimeInMillis,
+            threadPool::absoluteTimeInMillis);
     }
 
     /**
@@ -123,7 +139,7 @@ public void registerRepository(final PutRepositoryRequest request, final ActionL
 
         // Trying to create the new repository on master to make sure it works
         try {
-            closeRepository(createRepository(newRepositoryMetadata, typesRegistry));
+            closeRepository(createRepository(newRepositoryMetadata, typesRegistry), false);
         } catch (Exception e) {
             registrationListener.onFailure(e);
             return;
@@ -397,6 +413,27 @@ public Repository repository(String repositoryName) {
         throw new RepositoryMissingException(repositoryName);
     }
 
+    public List<RepositoryStatsSnapshot> repositoriesStats() {
+        List<RepositoryStatsSnapshot> archivedRepoStats = repositoriesStatsArchive.getArchivedStats();
+        List<RepositoryStatsSnapshot> activeRepoStats = getRepositoryStatsForActiveRepositories();
+
+        List<RepositoryStatsSnapshot> repositoriesStats = new ArrayList<>(archivedRepoStats);
+        repositoriesStats.addAll(activeRepoStats);
+        return repositoriesStats;
+    }
+
+    private List<RepositoryStatsSnapshot> getRepositoryStatsForActiveRepositories() {
+        return Stream.concat(repositories.values().stream(), internalRepositories.values().stream())
+            .filter(r -> r instanceof MeteredBlobStoreRepository)
+            .map(r -> (MeteredBlobStoreRepository) r)
+            .map(MeteredBlobStoreRepository::statsSnapshot)
+            .collect(Collectors.toList());
+    }
+
+    public List<RepositoryStatsSnapshot> clearRepositoriesStatsArchive() {
+        return repositoriesStatsArchive.clear();
+    }
+
     public void registerInternalRepository(String name, String type) {
         RepositoryMetadata metadata = new RepositoryMetadata(name, type, Settings.EMPTY);
         Repository repository = internalRepositories.computeIfAbsent(name, (n) -> {
@@ -423,8 +460,18 @@ public void unregisterInternalRepository(String name) {
 
     /** Closes the given repository. */
     private void closeRepository(Repository repository) {
+        closeRepository(repository, true);
+    }
+
+    private void closeRepository(Repository repository, boolean archiveStats) {
         logger.debug("closing repository [{}][{}]", repository.getMetadata().type(), repository.getMetadata().name());
         repository.close();
+        if (archiveStats && repository instanceof MeteredBlobStoreRepository) {
+            RepositoryStatsSnapshot stats = ((MeteredBlobStoreRepository) repository).statsSnapshot();
+            if (repositoriesStatsArchive.archive(stats) == false) {
+                logger.warn("Unable to archive the repository stats [{}] as the archive is full.", stats);
+            }
+        }
     }
 
     /**