Add repo throttle metrics to node stats api response

docs/changelog/96678.yaml

@@ -0,0 +1,6 @@
+pr: 96678
+summary: Add repo throttle metrics to node stats api response
+area: Snapshot/Restore
+type: feature
+issues:
+ - 89385

docs/reference/cluster/nodes-stats.asciidoc

@@ -84,6 +84,9 @@ using metrics.
  Process statistics, memory consumption, cpu usage, open
  file descriptors.
 
+ `repositories`::
+ Statistics about snapshot repositories.
+
  `thread_pool`::
  Statistics about each thread pool, including current size, queue and
  rejected tasks.
@@ -1672,6 +1675,31 @@ Total number of classes unloaded since the JVM started.
 =======
 ======
 
+[[cluster-nodes-stats-api-response-body-repositories]]
+`repositories`::
+(object)
+Statistics about snapshot repositories.
++
+.Properties of `repositories`
+[%collapsible%open]
+======
+`<repository_name>`::
+(object)
+Contains repository throttling statistics for the node.
++
+.Properties of `<repository_name>`
+[%collapsible%open]
+=======
+`total_read_throttled_time_nanos`::
+(integer)
+Total number of nanos which node had to wait during recovery.
+
+`total_write_throttled_time_nanos`::
+(integer)
+Total number of nanos which node had to wait during snapshotting.
+=======
+======
+
 [[cluster-nodes-stats-api-response-body-threadpool]]
 `thread_pool`::
 (object)

rest-api-spec/src/yamlRestTest/resources/rest-api-spec/test/nodes.stats/70_repository_throttling_stats.yml

@@ -0,0 +1,61 @@
+---
+"Repository throttling stats (no repository exists)":
+ - skip:
+ version: " - 8.8.99"
+ reason: "repository throttling stats was added in 8.9.0"
+ features: [arbitrary_key]
+
+ - do:
+ nodes.info: {}
+ - set:
+ nodes._arbitrary_key_: node_id
+
+ - do:
+ nodes.stats:
+ metric: [ repository ]
+
+ - is_true: nodes.$node_id.repositories
+ - match: { nodes.$node_id.repositories: {} }
+
+---
+"Repository throttling stats (some repositories exist)":
+ - skip:
+ version: " - 8.8.99"
+ reason: "repository throttling stats was added in 8.9.0"
+ features: [arbitrary_key]
+
+ - do:
+ nodes.info: {}
+ - set:
+ nodes._arbitrary_key_: node_id
+
+ - do:
+ snapshot.create_repository:
+ repository: test_repo_uuid_1
+ body:
+ type: fs
+ settings:
+ location: "test_repo_uuid_1_loc"
+
+ - do:
+ snapshot.create_repository:
+ repository: test_repo_uuid_2
+ body:
+ type: fs
+ settings:
+ location: "test_repo_uuid_2_loc"
+
+ - do:
+ nodes.stats:
+ metric: [ repository ]
+
+ - is_true: nodes.$node_id.repositories
+
+ - is_true: nodes.$node_id.repositories.test_repo_uuid_1
+ - gte: { nodes.$node_id.repositories.test_repo_uuid_1.total_read_throttled_time_nanos: 0 }
+ - gte: { nodes.$node_id.repositories.test_repo_uuid_1.total_write_throttled_time_nanos: 0 }
+
+ - is_true: nodes.$node_id.repositories.test_repo_uuid_2
+ - gte: { nodes.$node_id.repositories.test_repo_uuid_2.total_read_throttled_time_nanos: 0 }
+ - gte: { nodes.$node_id.repositories.test_repo_uuid_2.total_write_throttled_time_nanos: 0 }
+

server/src/internalClusterTest/java/org/elasticsearch/snapshots/RepositoryThrottlingStatsIT.java

@@ -0,0 +1,78 @@
+/*
+ * 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 and the Server Side Public License, v 1; you may not use this file except
+ * in compliance with, at your election, the Elastic License 2.0 or the Server
+ * Side Public License, v 1.
+ */
+
+package org.elasticsearch.snapshots;
+
+import org.elasticsearch.action.admin.cluster.node.stats.NodesStatsResponse;
+import org.elasticsearch.action.admin.cluster.snapshots.restore.RestoreSnapshotResponse;
+import org.elasticsearch.action.admin.indices.stats.IndexStats;
+import org.elasticsearch.action.admin.indices.stats.IndicesStatsResponse;
+import org.elasticsearch.action.admin.indices.stats.ShardStats;
+import org.elasticsearch.common.settings.Settings;
+import org.elasticsearch.common.unit.ByteSizeValue;
+import org.elasticsearch.repositories.RepositoriesStats;
+import org.elasticsearch.test.ESIntegTestCase;
+
+import java.util.Collections;
+
+import static org.hamcrest.Matchers.greaterThan;
+
+@ESIntegTestCase.ClusterScope(numDataNodes = 0, scope = ESIntegTestCase.Scope.TEST)
+public class RepositoryThrottlingStatsIT extends AbstractSnapshotIntegTestCase {
+
+ public void testRepositoryThrottlingStats() throws Exception {
+
+ logger.info("--> starting a node");
+ internalCluster().startNode();
+
+ logger.info("--> create index");
+ createIndexWithRandomDocs("test-idx", 100);
+
+ IndicesStatsResponse indicesStats = client().admin().indices().prepareStats("test-idx").get();
+ IndexStats indexStats = indicesStats.getIndex("test-idx");
+ long totalSizeInBytes = 0;
+ for (ShardStats shard : indexStats.getShards()) {
+ totalSizeInBytes += shard.getStats().getStore().getSizeInBytes();
+ }
+ logger.info("--> total shards size: {} bytes", totalSizeInBytes);
+
+ logger.info("--> create repository with really low snapshot/restore rate-limits");
+ createRepository(
+ "test-repo",
+ "fs",
+ Settings.builder()
+ .put("location", randomRepoPath())
+ .put("compress", false)
+ // set rate limits at ~25% of total size
+ .put("max_snapshot_bytes_per_sec", ByteSizeValue.ofBytes(totalSizeInBytes / 4))
+ .put("max_restore_bytes_per_sec", ByteSizeValue.ofBytes(totalSizeInBytes / 4))
+ );
+
+ logger.info("--> create snapshot");
+ createSnapshot("test-repo", "test-snap", Collections.singletonList("test-idx"));
+
+ logger.info("--> restore from snapshot");
+ RestoreSnapshotResponse restoreSnapshotResponse = clusterAdmin().prepareRestoreSnapshot("test-repo", "test-snap")
+ .setRenamePattern("test-")
+ .setRenameReplacement("test2-")
+ .setWaitForCompletion(true)
+ .execute()
+ .actionGet();
+ assertThat(restoreSnapshotResponse.getRestoreInfo().totalShards(), greaterThan(0));
+ assertDocCount("test-idx", 100);
+
+ logger.info("--> access repository throttling stats via _nodes/stats api");
+ NodesStatsResponse response = client().admin().cluster().prepareNodesStats().setRepositoryStats(true).get();
+ RepositoriesStats stats = response.getNodes().get(0).getRepositoriesStats();
+
+ assertTrue(stats.getRepositoryThrottlingStats().containsKey("test-repo"));
+ assertTrue(stats.getRepositoryThrottlingStats().get("test-repo").totalWriteThrottledNanos() > 0);
+ assertTrue(stats.getRepositoryThrottlingStats().get("test-repo").totalReadThrottledNanos() > 0);
+
+ }
+}

server/src/main/java/org/elasticsearch/TransportVersion.java

@@ -134,12 +134,13 @@ private static TransportVersion registerTransportVersion(int id, String uniqueId
  public static final TransportVersion V_8_500_008 = registerTransportVersion(8_500_008, "8884ab9d-94cd-4bac-aff8-01f2c394f47c");
  public static final TransportVersion V_8_500_009 = registerTransportVersion(8_500_009, "35091358-fd41-4106-a6e2-d2a1315494c1");
  public static final TransportVersion V_8_500_010 = registerTransportVersion(8_500_010, "9818C628-1EEC-439B-B943-468F61460675");
+ public static final TransportVersion V_8_500_011 = registerTransportVersion(8_500_011, "2209F28D-B52E-4BC4-9889-E780F291C32E");
 
  /**
  * Reference to the most recent transport version.
  * This should be the transport version with the highest id.
  */
- public static final TransportVersion CURRENT = findCurrent(V_8_500_010);
+ public static final TransportVersion CURRENT = findCurrent(V_8_500_011);
 
  /**
  * Reference to the earliest compatible transport version to this version of the codebase.

server/src/main/java/org/elasticsearch/action/admin/cluster/node/stats/NodeStats.java

@@ -8,6 +8,7 @@
 
 package org.elasticsearch.action.admin.cluster.node.stats;
 
+import org.elasticsearch.TransportVersion;
 import org.elasticsearch.action.support.nodes.BaseNodeResponse;
 import org.elasticsearch.cluster.node.DiscoveryNode;
 import org.elasticsearch.cluster.node.DiscoveryNodeRole;
@@ -27,6 +28,7 @@
 import org.elasticsearch.monitor.os.OsStats;
 import org.elasticsearch.monitor.process.ProcessStats;
 import org.elasticsearch.node.AdaptiveSelectionStats;
+import org.elasticsearch.repositories.RepositoriesStats;
 import org.elasticsearch.script.ScriptCacheStats;
 import org.elasticsearch.script.ScriptStats;
 import org.elasticsearch.threadpool.ThreadPoolStats;
@@ -92,6 +94,9 @@ public class NodeStats extends BaseNodeResponse implements ChunkedToXContent {
  @Nullable
  private final IndexingPressureStats indexingPressureStats;
 
+ @Nullable
+ private final RepositoriesStats repositoriesStats;
+
  public NodeStats(StreamInput in) throws IOException {
  super(in);
  timestamp = in.readVLong();
@@ -112,6 +117,11 @@ public NodeStats(StreamInput in) throws IOException {
  ingestStats = in.readOptionalWriteable(IngestStats::read);
  adaptiveSelectionStats = in.readOptionalWriteable(AdaptiveSelectionStats::new);
  indexingPressureStats = in.readOptionalWriteable(IndexingPressureStats::new);
+ if (in.getTransportVersion().onOrAfter(TransportVersion.V_8_500_011)) {
+ repositoriesStats = in.readOptionalWriteable(RepositoriesStats::new);
+ } else {
+ repositoriesStats = null;
+ }
  }
 
  public NodeStats(
@@ -131,7 +141,8 @@ public NodeStats(
  @Nullable IngestStats ingestStats,
  @Nullable AdaptiveSelectionStats adaptiveSelectionStats,
  @Nullable ScriptCacheStats scriptCacheStats,
- @Nullable IndexingPressureStats indexingPressureStats
+ @Nullable IndexingPressureStats indexingPressureStats,
+ @Nullable RepositoriesStats repositoriesStats
  ) {
  super(node);
  this.timestamp = timestamp;
@@ -150,6 +161,7 @@ public NodeStats(
  this.adaptiveSelectionStats = adaptiveSelectionStats;
  this.scriptCacheStats = scriptCacheStats;
  this.indexingPressureStats = indexingPressureStats;
+ this.repositoriesStats = repositoriesStats;
  }
 
  public long getTimestamp() {
@@ -254,6 +266,11 @@ public IndexingPressureStats getIndexingPressureStats() {
  return indexingPressureStats;
  }
 
+ @Nullable
+ public RepositoriesStats getRepositoriesStats() {
+ return repositoriesStats;
+ }
+
  @Override
  public void writeTo(StreamOutput out) throws IOException {
  super.writeTo(out);
@@ -277,6 +294,9 @@ public void writeTo(StreamOutput out) throws IOException {
  out.writeOptionalWriteable(ingestStats);
  out.writeOptionalWriteable(adaptiveSelectionStats);
  out.writeOptionalWriteable(indexingPressureStats);
+ if (out.getTransportVersion().onOrAfter(TransportVersion.V_8_500_011)) {
+ out.writeOptionalWriteable(repositoriesStats);
+ }
  }
 
  @Override
@@ -309,12 +329,7 @@ public Iterator<? extends ToXContent> toXContentChunked(ToXContent.Params outerP
 
  ifPresent(getIndices()).toXContentChunked(outerParams),
 
- Iterators.single((builder, params) -> {
- ifPresent(getOs()).toXContent(builder, params);
- ifPresent(getProcess()).toXContent(builder, params);
- ifPresent(getJvm()).toXContent(builder, params);
- return builder;
- }),
+ singleChunk(ifPresent(getOs()), ifPresent(getProcess()), ifPresent(getJvm())),
 
  ifPresent(getThreadPool()).toXContentChunked(outerParams),
  singleChunk(ifPresent(getFs())),
@@ -326,7 +341,8 @@ public Iterator<? extends ToXContent> toXContentChunked(ToXContent.Params outerP
  ifPresent(getIngestStats()).toXContentChunked(outerParams),
  singleChunk(ifPresent(getAdaptiveSelectionStats())),
  ifPresent(getScriptCacheStats()).toXContentChunked(outerParams),
- singleChunk(ifPresent(getIndexingPressureStats()))
+ singleChunk(ifPresent(getIndexingPressureStats())),
+ singleChunk(ifPresent(getRepositoriesStats()))
  );
  }
 

server/src/main/java/org/elasticsearch/action/admin/cluster/node/stats/NodesStatsRequest.java

@@ -178,7 +178,8 @@ public enum Metric {
  INGEST("ingest"),
  ADAPTIVE_SELECTION("adaptive_selection"),
  SCRIPT_CACHE("script_cache"),
- INDEXING_PRESSURE("indexing_pressure");
+ INDEXING_PRESSURE("indexing_pressure"),
+ REPOSITORY("repository");
 
  private String metricName;
 

server/src/main/java/org/elasticsearch/action/admin/cluster/node/stats/NodesStatsRequestBuilder.java

@@ -153,6 +153,11 @@ public NodesStatsRequestBuilder setIndexingPressure(boolean indexingPressure) {
  return this;
  }
 
+ public NodesStatsRequestBuilder setRepositoryStats(boolean repositoryStats) {
+ addOrRemoveMetric(repositoryStats, NodesStatsRequest.Metric.REPOSITORY);
+ return this;
+ }
+
  /**
  * Helper method for adding metrics to a request
  */

server/src/main/java/org/elasticsearch/action/admin/cluster/node/stats/TransportNodesStatsAction.java

@@ -95,7 +95,8 @@ protected NodeStats nodeOperation(NodeStatsRequest nodeStatsRequest, Task task)
  NodesStatsRequest.Metric.INGEST.containedIn(metrics),
  NodesStatsRequest.Metric.ADAPTIVE_SELECTION.containedIn(metrics),
  NodesStatsRequest.Metric.SCRIPT_CACHE.containedIn(metrics),
- NodesStatsRequest.Metric.INDEXING_PRESSURE.containedIn(metrics)
+ NodesStatsRequest.Metric.INDEXING_PRESSURE.containedIn(metrics),
+ NodesStatsRequest.Metric.REPOSITORY.containedIn(metrics)
  );
  }
 

server/src/main/java/org/elasticsearch/action/admin/cluster/stats/TransportClusterStatsAction.java

@@ -196,6 +196,7 @@ protected ClusterStatsNodeResponse nodeOperation(ClusterStatsNodeRequest nodeReq
  true,
  false,
  false,
+ false,
  false
  );
  List<ShardStats> shardsStats = new ArrayList<>();

server/src/main/java/org/elasticsearch/health/node/LocalHealthMonitor.java

@@ -466,6 +466,7 @@ private DiskUsage getDiskUsage() {
  false,
  false,
  false,
+ false,
  false
  );
  return DiskUsage.findLeastAvailablePath(nodeStats);

server/src/main/java/org/elasticsearch/node/Node.java

@@ -953,7 +953,8 @@ protected Node(
  responseCollectorService,
  searchTransportService,
  indexingLimits,
- searchModule.getValuesSourceRegistry().getUsageService()
+ searchModule.getValuesSourceRegistry().getUsageService(),
+ repositoryService
  );
 
  final SearchService searchService = newSearchService(