Skip to content

Fix SnapshotMetricsIT race conditions #132780

New issue

Have a question about this project? Sign up for a free GitHub account to open an issue and contact its maintainers and the community.

Sign up for GitHub

By clicking “Sign up for GitHub”, you agree to our terms of service and privacy statement. We’ll occasionally send you account related emails.

Already on GitHub? Sign in to your account

Merged
nicktindall merged 15 commits into from
Aug 20, 2025
Merged

Fix SnapshotMetricsIT race conditions #132780

Show file tree
Hide file tree
Changes from all commits
Commits
Show all changes
15 commits
Select commit Hold shift + click to select a range
d73d299
Fix SnapshotMetricsIT race conditions
nicktindall Aug 13, 2025
2c66600
Increase throttling
nicktindall Aug 13, 2025
f034e1d
Test throttle metrics separately
nicktindall Aug 13, 2025
23497d4
Increase byte_per_sec, don't verify on create/update repo, add loggin…
nicktindall Aug 14, 2025
8a71043
Merge remote-tracking branch 'origin/main' into alt_fix_snapshot_metr…
nicktindall Aug 14, 2025
e7722b2
Merge remote-tracking branch 'origin/main' into alt_fix_snapshot_metr…
nicktindall Aug 14, 2025
62f1935
Don't verify repository when turning off restore throttling
nicktindall Aug 14, 2025
ca43178
Fix the last(?) race condition
nicktindall Aug 15, 2025
db4d5c6
Uuurgh
nicktindall Aug 15, 2025
a57ddb4
Merge remote-tracking branch 'origin/main' into alt_fix_snapshot_metr…
nicktindall Aug 19, 2025
0a0b09a
Remove remnants of project_id label
nicktindall Aug 19, 2025
a397141
Increase size of index to ensure upload takes time
nicktindall Aug 19, 2025
48ed1e5
Try and fix upload duration flappiness
nicktindall Aug 19, 2025
4dbe2ce
Use ofKb instead of ofBytes
nicktindall Aug 19, 2025
37a4c91
Merge remote-tracking branch 'origin/main' into alt_fix_snapshot_metr…
nicktindall Aug 19, 2025
File filter

Filter by extension

Filter by extension
Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
3 changes: 0 additions & 3 deletions muted-tests.yml
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -471,9 +471,6 @@ tests:
- class: org.elasticsearch.xpack.ml.integration.RevertModelSnapshotIT
method: testRevertModelSnapshot
issue: https://github.com/elastic/elasticsearch/issues/132733
- class: org.elasticsearch.repositories.SnapshotMetricsIT
method: testSnapshotAPMMetrics
issue: https://github.com/elastic/elasticsearch/issues/132731
- class: org.elasticsearch.xpack.esql.qa.single_node.GenerativeForkIT
method: test {csv-spec:lookup-join.MvJoinKeyFromRowExpanded}
issue: https://github.com/elastic/elasticsearch/issues/132778
Expand Down
238 changes: 160 additions & 78 deletions server/src/internalClusterTest/java/org/elasticsearch/repositories/SnapshotMetricsIT.java
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -9,12 +9,11 @@

package org.elasticsearch.repositories;

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.action.ActionFuture;
import org.elasticsearch.action.admin.cluster.snapshots.create.CreateSnapshotResponse;
import org.elasticsearch.action.admin.cluster.snapshots.restore.RestoreSnapshotResponse;
import org.elasticsearch.cluster.SnapshotsInProgress;
import org.elasticsearch.cluster.metadata.IndexMetadata;
import org.elasticsearch.cluster.metadata.ProjectId;
import org.elasticsearch.cluster.service.ClusterService;
import org.elasticsearch.common.settings.Settings;
import org.elasticsearch.common.unit.ByteSizeValue;
Expand Down Expand Up @@ -83,47 +82,38 @@ public void testSnapshotAPMMetrics() throws Exception {
final int numReplicas = randomIntBetween(0, 1);
createIndex(indexName, numShards, numReplicas);

indexRandom(true, indexName, randomIntBetween(100, 300));

final IndicesStatsResponse indicesStats = indicesAdmin().prepareStats(indexName).get();
final IndexStats indexStats = indicesStats.getIndex(indexName);
long totalSizeInBytes = 0;
for (ShardStats shard : indexStats.getShards()) {
totalSizeInBytes += shard.getStats().getStore().sizeInBytes();
}
logger.info("--> total shards size: {} bytes", totalSizeInBytes);
indexRandom(true, indexName, randomIntBetween(3000, 5000));

final String repositoryName = randomIdentifier();

// we want to ensure some throttling, but not so much that it makes the test excessively slow.
final int shardSizeMultipleToEnsureThrottling = 2;
createRepository(
repositoryName,
"mock",
randomRepositorySettings().put(
BlobStoreRepository.MAX_SNAPSHOT_BYTES_PER_SEC.getKey(),
ByteSizeValue.ofBytes(totalSizeInBytes * shardSizeMultipleToEnsureThrottling)
)
.put(
BlobStoreRepository.MAX_RESTORE_BYTES_PER_SEC.getKey(),
ByteSizeValue.ofBytes(totalSizeInBytes * shardSizeMultipleToEnsureThrottling)
)
Settings.builder()
.put(randomRepositorySettings().build())
// Making chunk size small and adding throttling increases the likelihood of upload duration being non-zero
.put("chunk_size", ByteSizeValue.ofKb(1))
.put(BlobStoreRepository.MAX_SNAPSHOT_BYTES_PER_SEC.getKey(), ByteSizeValue.ofMb(1))
.put(BlobStoreRepository.MAX_RESTORE_BYTES_PER_SEC.getKey(), ByteSizeValue.ofMb(1))
);

// Block the snapshot to test "snapshot shards in progress"
blockAllDataNodes(repositoryName);
final String snapshotName = randomIdentifier();
final long beforeCreateSnapshotNanos = System.nanoTime();
final ActionFuture<CreateSnapshotResponse> snapshotFuture;
try {
clusterAdmin().prepareCreateSnapshot(TEST_REQUEST_TIMEOUT, repositoryName, snapshotName)
snapshotFuture = clusterAdmin().prepareCreateSnapshot(TEST_REQUEST_TIMEOUT, repositoryName, snapshotName)
.setIndices(indexName)
.setWaitForCompletion(false)
.get();

waitForBlockOnAnyDataNode(repositoryName);
collectMetrics();
assertShardsInProgressMetricIs(hasItem(greaterThan(0L)));
assertThat(getTotalClusterLongCounterValue(SnapshotMetrics.SNAPSHOTS_STARTED), equalTo(1L));
.setWaitForCompletion(true)
.execute();

// We are able to wait for either the creation to complete (`wait_for_completion=false`), or the snapshot to complete
// (`wait_for_completion=true`), but not both. To know when the creation listeners complete, we must assertBusy
assertBusy(() -> {
collectMetrics();
assertThat(getTotalClusterLongCounterValue(SnapshotMetrics.SNAPSHOTS_STARTED), equalTo(1L));
assertShardsInProgressMetricIs(hasItem(greaterThan(0L)));
});
assertThat(getTotalClusterLongCounterValue(SnapshotMetrics.SNAPSHOTS_COMPLETED), equalTo(0L));
assertThat(getTotalClusterLongCounterValue(SnapshotMetrics.SNAPSHOT_SHARDS_STARTED), greaterThan(0L));
assertThat(getTotalClusterLongCounterValue(SnapshotMetrics.SNAPSHOT_SHARDS_COMPLETED), equalTo(0L));
Expand All @@ -132,15 +122,13 @@ public void testSnapshotAPMMetrics() throws Exception {
}

// wait for snapshot to finish to test the other metrics
awaitNumberOfSnapshotsInProgress(0);
safeGet(snapshotFuture);
Copy link
Contributor Author

@nicktindall nicktindall Aug 13, 2025 •
edited
Loading

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

I think if we wait for the response to be received that should mean the metrics should be written? (note waitForCompletion=true above now)

I'm running in a loop to try and confirm that now

Copy link
Contributor Author

@nicktindall nicktindall Aug 13, 2025 •
edited
Loading

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

This introduced a source of flakiness in the tension between the throttling and safeGet. If the setting was too high, we exceeded the safeGet timeout, if the setting was too low, we didn't see any throttling.

For this reason I've moved out the test for the throttling metrics to another test. It has more complexity to ensure we see throttling and we don't run too long.

final TimeValue snapshotElapsedTime = TimeValue.timeValueNanos(System.nanoTime() - beforeCreateSnapshotNanos);
collectMetrics();

// sanity check blobs, bytes and throttling metrics
assertThat(getTotalClusterLongCounterValue(SnapshotMetrics.SNAPSHOT_BLOBS_UPLOADED), greaterThan(0L));
assertThat(getTotalClusterLongCounterValue(SnapshotMetrics.SNAPSHOT_BYTES_UPLOADED), greaterThan(0L));
assertThat(getTotalClusterLongCounterValue(SnapshotMetrics.SNAPSHOT_CREATE_THROTTLE_DURATION), greaterThan(0L));
assertThat(getTotalClusterLongCounterValue(SnapshotMetrics.SNAPSHOT_RESTORE_THROTTLE_DURATION), equalTo(0L));

assertThat(getTotalClusterLongCounterValue(SnapshotMetrics.SNAPSHOTS_STARTED), equalTo(1L));
assertThat(getTotalClusterLongCounterValue(SnapshotMetrics.SNAPSHOTS_COMPLETED), equalTo(1L));
Expand All @@ -165,37 +153,14 @@ public void testSnapshotAPMMetrics() throws Exception {
getTotalClusterLongCounterValue(SnapshotMetrics.SNAPSHOT_UPLOAD_DURATION),
allOf(greaterThan(0L), lessThan(upperBoundTimeSpentOnSnapshotThingsMillis))
);
assertThat(
getTotalClusterLongCounterValue(SnapshotMetrics.SNAPSHOT_UPLOAD_READ_DURATION),
allOf(greaterThan(0L), lessThan(upperBoundTimeSpentOnSnapshotThingsMillis))
);

assertThat(getTotalClusterLongCounterValue(SnapshotMetrics.SNAPSHOT_SHARDS_STARTED), equalTo((long) numShards));
assertThat(getTotalClusterLongCounterValue(SnapshotMetrics.SNAPSHOT_SHARDS_COMPLETED), equalTo((long) numShards));

assertShardsInProgressMetricIs(everyItem(equalTo(0L)));

// Restore the snapshot
clusterAdmin().prepareRestoreSnapshot(TEST_REQUEST_TIMEOUT, repositoryName, snapshotName)
.setIndices(indexName)
.setWaitForCompletion(true)
.setRenamePattern("(.+)")
.setRenameReplacement("restored-$1")
.get();
collectMetrics();

// assert we throttled on restore
assertThat(getTotalClusterLongCounterValue(SnapshotMetrics.SNAPSHOT_RESTORE_THROTTLE_DURATION), greaterThan(0L));

// assert appropriate attributes are present
final Map<String, Object> expectedAttrs = Map.of(
"project_id",
ProjectId.DEFAULT.id(),
"repo_name",
repositoryName,
"repo_type",
"mock"
);
final Map<String, Object> expectedAttrs = Map.of("repo_name", repositoryName, "repo_type", "mock");
final Map<String, Object> expectedAttrsWithShardStage = Maps.copyMapWithAddedEntry(
expectedAttrs,
"stage",
Expand All @@ -215,14 +180,121 @@ public void testSnapshotAPMMetrics() throws Exception {
assertMetricsHaveAttributes(InstrumentType.LONG_COUNTER, SnapshotMetrics.SNAPSHOT_SHARDS_COMPLETED, expectedAttrsWithShardStage);
assertMetricsHaveAttributes(InstrumentType.DOUBLE_HISTOGRAM, SnapshotMetrics.SNAPSHOT_SHARDS_DURATION, expectedAttrsWithShardStage);

assertMetricsHaveAttributes(InstrumentType.LONG_COUNTER, SnapshotMetrics.SNAPSHOT_RESTORE_THROTTLE_DURATION, expectedAttrs);
assertMetricsHaveAttributes(InstrumentType.LONG_COUNTER, SnapshotMetrics.SNAPSHOT_CREATE_THROTTLE_DURATION, expectedAttrs);
assertMetricsHaveAttributes(InstrumentType.LONG_COUNTER, SnapshotMetrics.SNAPSHOT_UPLOAD_READ_DURATION, expectedAttrs);
assertMetricsHaveAttributes(InstrumentType.LONG_COUNTER, SnapshotMetrics.SNAPSHOT_UPLOAD_DURATION, expectedAttrs);
assertMetricsHaveAttributes(InstrumentType.LONG_COUNTER, SnapshotMetrics.SNAPSHOT_BYTES_UPLOADED, expectedAttrs);
assertMetricsHaveAttributes(InstrumentType.LONG_COUNTER, SnapshotMetrics.SNAPSHOT_BLOBS_UPLOADED, expectedAttrs);
}

public void testThrottlingMetrics() throws Exception {
final String indexName = randomIdentifier();
final int numShards = randomIntBetween(1, 10);
final int numReplicas = randomIntBetween(0, 1);
createIndex(indexName, numShards, numReplicas);
indexRandom(true, indexName, randomIntBetween(100, 120));

// Create a repository with restrictive throttling settings
final String repositoryName = randomIdentifier();
final Settings.Builder repositorySettings = randomRepositorySettings().put(
BlobStoreRepository.MAX_SNAPSHOT_BYTES_PER_SEC.getKey(),
ByteSizeValue.ofKb(2)
)
.put(BlobStoreRepository.MAX_RESTORE_BYTES_PER_SEC.getKey(), ByteSizeValue.ofKb(2))
// Small chunk size ensures we don't get stuck throttling for too long
.put("chunk_size", ByteSizeValue.ofBytes(100));
createRepository(repositoryName, "mock", repositorySettings, false);

final String snapshotName = randomIdentifier();
final ActionFuture<CreateSnapshotResponse> snapshotFuture;

// Kick off a snapshot
final long snapshotStartTime = System.currentTimeMillis();
snapshotFuture = clusterAdmin().prepareCreateSnapshot(TEST_REQUEST_TIMEOUT, repositoryName, snapshotName)
.setIndices(indexName)
.setWaitForCompletion(true)
.execute();

// Poll until we see some throttling occurring
final long snap_ts0 = System.currentTimeMillis();
assertBusy(() -> {
collectMetrics();
assertThat(getTotalClusterLongCounterValue(SnapshotMetrics.SNAPSHOT_CREATE_THROTTLE_DURATION), greaterThan(0L));
});
assertThat(getTotalClusterLongCounterValue(SnapshotMetrics.SNAPSHOT_RESTORE_THROTTLE_DURATION), equalTo(0L));

// Remove create throttling
final long snap_ts1 = System.currentTimeMillis();
createRepository(
repositoryName,
"mock",
repositorySettings.put(BlobStoreRepository.MAX_SNAPSHOT_BYTES_PER_SEC.getKey(), ByteSizeValue.ZERO),
false
);
final long snap_ts2 = System.currentTimeMillis();

// wait for the snapshot to finish
safeGet(snapshotFuture);
final long snap_ts3 = System.currentTimeMillis();

logger.info(
"saw throttling in [{}] remove throttling took [{}], snapshot took [{}]",
TimeValue.timeValueMillis(snap_ts1 - snap_ts0),
TimeValue.timeValueMillis(snap_ts2 - snap_ts1),
TimeValue.timeValueMillis(snap_ts3 - snap_ts2)
);

// Work out the maximum amount of concurrency per node
final ThreadPool tp = internalCluster().getDataNodeInstance(ThreadPool.class);
final int snapshotThreadPoolSize = tp.info(ThreadPool.Names.SNAPSHOT).getMax();
final int maximumPerNodeConcurrency = Math.max(snapshotThreadPoolSize, numShards);

// we should also have incurred some read duration due to the throttling
final long upperBoundTimeSpentOnSnapshotThingsMillis = internalCluster().numDataNodes() * maximumPerNodeConcurrency * (System
.currentTimeMillis() - snapshotStartTime);
assertThat(
getTotalClusterLongCounterValue(SnapshotMetrics.SNAPSHOT_UPLOAD_READ_DURATION),
allOf(greaterThan(0L), lessThan(upperBoundTimeSpentOnSnapshotThingsMillis))
);

// Restore the snapshot
final long restore_ts0 = System.currentTimeMillis();
ActionFuture<RestoreSnapshotResponse> restoreFuture = clusterAdmin().prepareRestoreSnapshot(
TEST_REQUEST_TIMEOUT,
repositoryName,
snapshotName
).setIndices(indexName).setWaitForCompletion(true).setRenamePattern("(.+)").setRenameReplacement("restored-$1").execute();

final long restore_ts1 = System.currentTimeMillis();
// assert we throttled on restore
assertBusy(() -> {
collectMetrics();
assertThat(getTotalClusterLongCounterValue(SnapshotMetrics.SNAPSHOT_RESTORE_THROTTLE_DURATION), greaterThan(0L));
});

final long restore_ts2 = System.currentTimeMillis();
// Remove restore throttling
createRepository(
repositoryName,
"mock",
repositorySettings.put(BlobStoreRepository.MAX_RESTORE_BYTES_PER_SEC.getKey(), ByteSizeValue.ZERO),
false
);
safeGet(restoreFuture);
final long restore_ts3 = System.currentTimeMillis();

logger.info(
"saw throttling in [{}] remove throttling took [{}], restore took [{}]",
TimeValue.timeValueMillis(restore_ts1 - restore_ts0),
TimeValue.timeValueMillis(restore_ts2 - restore_ts1),
TimeValue.timeValueMillis(restore_ts3 - restore_ts2)
);

// assert appropriate attributes are present
final Map<String, Object> expectedAttrs = Map.of("repo_name", repositoryName, "repo_type", "mock");
assertMetricsHaveAttributes(InstrumentType.LONG_COUNTER, SnapshotMetrics.SNAPSHOT_UPLOAD_READ_DURATION, expectedAttrs);
assertMetricsHaveAttributes(InstrumentType.LONG_COUNTER, SnapshotMetrics.SNAPSHOT_RESTORE_THROTTLE_DURATION, expectedAttrs);
assertMetricsHaveAttributes(InstrumentType.LONG_COUNTER, SnapshotMetrics.SNAPSHOT_CREATE_THROTTLE_DURATION, expectedAttrs);
}

public void testByStateCounts_InitAndQueuedShards() throws Exception {
final String indexName = randomIdentifier();
final int numShards = randomIntBetween(2, 10);
Expand All @@ -237,11 +309,13 @@ public void testByStateCounts_InitAndQueuedShards() throws Exception {
blockAllDataNodes(repositoryName);

final String snapshotName = randomIdentifier();
final ActionFuture<CreateSnapshotResponse> firstSnapshotFuture;
final ActionFuture<CreateSnapshotResponse> secondSnapshotFuture;
try {
clusterAdmin().prepareCreateSnapshot(TEST_REQUEST_TIMEOUT, repositoryName, snapshotName)
firstSnapshotFuture = clusterAdmin().prepareCreateSnapshot(TEST_REQUEST_TIMEOUT, repositoryName, snapshotName)
.setIndices(indexName)
.setWaitForCompletion(false)
.get();
.setWaitForCompletion(true)
.execute();

waitForBlockOnAnyDataNode(repositoryName);

Expand All @@ -252,10 +326,12 @@ public void testByStateCounts_InitAndQueuedShards() throws Exception {
assertThat(snapshotStates.get(SnapshotsInProgress.State.STARTED), equalTo(1L));

// Queue up another snapshot
clusterAdmin().prepareCreateSnapshot(TEST_REQUEST_TIMEOUT, repositoryName, randomIdentifier())
secondSnapshotFuture = clusterAdmin().prepareCreateSnapshot(TEST_REQUEST_TIMEOUT, repositoryName, randomIdentifier())
.setIndices(indexName)
.setWaitForCompletion(false)
.get();
.setWaitForCompletion(true)
.execute();

awaitNumberOfSnapshotsInProgress(2);

// Should be {numShards} in QUEUED and INIT states, and 2 STARTED snapshots
shardStates = getShardStates();
Expand All @@ -268,7 +344,8 @@ public void testByStateCounts_InitAndQueuedShards() throws Exception {
}

// All statuses should return to zero when the snapshots complete
awaitNumberOfSnapshotsInProgress(0);
safeGet(firstSnapshotFuture);
safeGet(secondSnapshotFuture);
getShardStates().forEach((key, value) -> assertThat(value, equalTo(0L)));
getSnapshotStates().forEach((key, value) -> assertThat(value, equalTo(0L)));

Expand Down Expand Up @@ -309,12 +386,13 @@ public void testByStateCounts_PausedForRemovalShards() throws Exception {
blockNodeOnAnyFiles(repositoryName, nodeForRemoval);

final ClusterService clusterService = internalCluster().getCurrentMasterNodeInstance(ClusterService.class);
final ActionFuture<CreateSnapshotResponse> snapshotFuture;
try {
// Kick off a snapshot
clusterAdmin().prepareCreateSnapshot(TEST_REQUEST_TIMEOUT, repositoryName, randomIdentifier())
snapshotFuture = clusterAdmin().prepareCreateSnapshot(TEST_REQUEST_TIMEOUT, repositoryName, randomIdentifier())
.setIndices(indexName)
.setWaitForCompletion(false)
.get();
.setWaitForCompletion(true)
.execute();

// Wait till we're blocked
waitForBlock(nodeForRemoval, repositoryName);
Expand All @@ -337,7 +415,7 @@ public void testByStateCounts_PausedForRemovalShards() throws Exception {
clearShutdownMetadata(clusterService);

// All statuses should return to zero when the snapshot completes
awaitNumberOfSnapshotsInProgress(0);
safeGet(snapshotFuture);
getShardStates().forEach((key, value) -> assertThat(value, equalTo(0L)));
getSnapshotStates().forEach((key, value) -> assertThat(value, equalTo(0L)));

Expand Down Expand Up @@ -395,10 +473,14 @@ public void testByStateCounts_WaitingShards() throws Exception {
safeAwait(handoffRequestBarrier);

// Kick off a snapshot
clusterAdmin().prepareCreateSnapshot(TEST_REQUEST_TIMEOUT, repositoryName, randomIdentifier())
.setIndices(indexName)
.setWaitForCompletion(false)
.get();
final ActionFuture<CreateSnapshotResponse> snapshotFuture = clusterAdmin().prepareCreateSnapshot(
TEST_REQUEST_TIMEOUT,
repositoryName,
randomIdentifier()
).setIndices(indexName).setWaitForCompletion(true).execute();

// Wait for the snapshot to start
awaitNumberOfSnapshotsInProgress(1);

// Wait till we see a shard in WAITING state
createSnapshotInStateListener(clusterService(), repositoryName, indexName, 1, SnapshotsInProgress.ShardState.WAITING);
Expand All @@ -413,7 +495,7 @@ public void testByStateCounts_WaitingShards() throws Exception {
safeAwait(handoffRequestBarrier);

// All statuses should return to zero when the snapshot completes
awaitNumberOfSnapshotsInProgress(0);
safeGet(snapshotFuture);
getShardStates().forEach((key, value) -> assertThat(value, equalTo(0L)));
getSnapshotStates().forEach((key, value) -> assertThat(value, equalTo(0L)));

Expand Down