Skip to content

Commit 916689e

Browse files
tlrxtlrx
committed
skip cluster state updates
1 parent 13a7f69 commit 916689e

File tree

4 files changed

+182
-71
lines changed

4 files changed

+182
-71
lines changed

‎server/src/main/java/org/elasticsearch/cluster/SnapshotDeletionsPending.java

Lines changed: 13 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -144,6 +144,19 @@ public SnapshotDeletionsPending withRemovedSnapshots(List<SnapshotId> snapshotId
144144
}
145145
}
146146

147+
@Override
148+
public boolean equals(Object o) {
149+
if (this == o) return true;
150+
if (o == null || getClass() != o.getClass()) return false;
151+
SnapshotDeletionsPending that = (SnapshotDeletionsPending) o;
152+
return Objects.equals(entries, that.entries);
153+
}
154+
155+
@Override
156+
public int hashCode() {
157+
return Objects.hash(entries);
158+
}
159+
147160
@Override
148161
public String toString() {
149162
return "SnapshotDeletionsPending[" + entries + ']';

‎server/src/main/java/org/elasticsearch/cluster/metadata/MetadataDeleteIndexService.java

Lines changed: 6 additions & 6 deletions
Original file line numberDiff line numberDiff line change
@@ -223,8 +223,8 @@ private SnapshotDeletionsPending updateSnapshotDeletionsPending(
223223
.collect(Collectors.toUnmodifiableSet());
224224

225225
final RepositoriesMetadata repositories = state.metadata().custom(RepositoriesMetadata.TYPE);
226-
// used to deduplicate snapshots that were used by multiple deleted indices
227-
final Map<SnapshotId, RepositoryMetadata> snapshotsWithRepository = new HashMap<>();
226+
// also used to deduplicate snapshots that were used by multiple deleted indices
227+
final Map<SnapshotId, Tuple<String, String>> snapshotsWithRepository = new HashMap<>();
228228
// also used to log a warning for snapshots with unknown repository
229229
final Map<SnapshotId, Tuple<String, String>> snapshotsWithoutRepository = new HashMap<>();
230230

@@ -235,7 +235,7 @@ private SnapshotDeletionsPending updateSnapshotDeletionsPending(
235235
String repositoryName = deletedIndexSettings.get(SEARCHABLE_SNAPSHOTS_REPOSITORY_NAME_SETTING_KEY);
236236
Optional<RepositoryMetadata> repository = findRepositoryForPendingDeletion(repositories, repositoryName, repositoryUuid);
237237
if (repository.isPresent()) {
238-
snapshotsWithRepository.putIfAbsent(snapshotId, repository.get());
238+
snapshotsWithRepository.putIfAbsent(snapshotId, Tuple.tuple(repositoryName, repositoryUuid));
239239
} else {
240240
snapshotsWithoutRepository.putIfAbsent(snapshotId, Tuple.tuple(repositoryName, repositoryUuid));
241241
}
@@ -259,9 +259,9 @@ private SnapshotDeletionsPending updateSnapshotDeletionsPending(
259259
);
260260

261261
final long timestamp = Instant.now().toEpochMilli();
262-
for (Map.Entry<SnapshotId, RepositoryMetadata> entry : snapshotsWithRepository.entrySet()) {
263-
logger.info("snapshot [{}:{}] added to the list of snapshots pending deletion", entry.getValue().name(), entry.getKey());
264-
builder.add(entry.getValue().name(), entry.getValue().uuid(), entry.getKey(), timestamp);
262+
for (Map.Entry<SnapshotId, Tuple<String, String>> entry : snapshotsWithRepository.entrySet()) {
263+
logger.debug("snapshot [{}:{}] added to the list of snapshots pending deletion", entry.getValue().v1(), entry.getKey());
264+
builder.add(entry.getValue().v1(), entry.getValue().v2(), entry.getKey(), timestamp);
265265
}
266266
for (Map.Entry<SnapshotId, Tuple<String, String>> entry : snapshotsWithoutRepository.entrySet()) {
267267
// TODO also log that it will stay as pending for a given time/attempts and then be removed?

‎server/src/main/java/org/elasticsearch/snapshots/SnapshotsService.java

Lines changed: 162 additions & 64 deletions
Original file line numberDiff line numberDiff line change
@@ -992,17 +992,19 @@ public void applyClusterState(ClusterChangedEvent event) {
992992
newMaster || removedNodesCleanupNeeded(snapshotsInProgress, event.nodesDelta().removedNodes()),
993993
event.routingTableChanged() && waitingShardsStartedOrUnassigned(snapshotsInProgress, event)
994994
);
995-
triggerSnapshotsPendingDeletions(event.state());
996-
} else if (snapshotCompletionListeners.isEmpty() == false) {
997-
// We have snapshot listeners but are not the master any more. Fail all waiting listeners except for those that already
998-
// have their snapshots finalizing (those that are already finalizing will fail on their own from to update the cluster
999-
// state).
1000-
for (Snapshot snapshot : Set.copyOf(snapshotCompletionListeners.keySet())) {
1001-
if (endingSnapshots.add(snapshot)) {
1002-
failSnapshotCompletionListeners(snapshot, new SnapshotException(snapshot, "no longer master"));
995+
} else {
996+
if (snapshotCompletionListeners.isEmpty() == false) {
997+
// We have snapshot listeners but are not the master any more. Fail all waiting listeners except for those that already
998+
// have their snapshots finalizing (those that are already finalizing will fail on their own from to update the cluster
999+
// state).
1000+
for (Snapshot snapshot : Set.copyOf(snapshotCompletionListeners.keySet())) {
1001+
if (endingSnapshots.add(snapshot)) {
1002+
failSnapshotCompletionListeners(snapshot, new SnapshotException(snapshot, "no longer master"));
1003+
}
10031004
}
10041005
}
10051006
}
1007+
triggerSnapshotsPendingDeletions(event);
10061008
} catch (Exception e) {
10071009
assert false : new AssertionError(e);
10081010
logger.warn("Failed to update snapshot state ", e);
@@ -1295,6 +1297,13 @@ public void clusterStateProcessed(String source, ClusterState oldState, ClusterS
12951297
*/
12961298
private final Set<SnapshotId> ongoingSnapshotsDeletions = ConcurrentCollections.newConcurrentSet();
12971299

1300+
/**
1301+
* Set of pending snapshots deletions whose deletion is conflicting with on-going restores, clones or repository statuses
1302+
*/
1303+
private final Set<SnapshotId> pendingDeletionsWithConflictingRestores = ConcurrentCollections.newConcurrentSet();
1304+
private final Set<SnapshotId> pendingDeletionsWithConflictingClones = ConcurrentCollections.newConcurrentSet();
1305+
private final Set<SnapshotId> pendingDeletionsWithConflictingRepos = ConcurrentCollections.newConcurrentSet();
1306+
12981307
/**
12991308
* Find snapshots to delete in the the cluster state and triggers explicit snapshot delete requests. This method attempts to detect
13001309
* conflicting situations where triggering the snapshot deletion would likely fail due to a concurrent snapshot operation. In such
@@ -1307,77 +1316,165 @@ public void clusterStateProcessed(String source, ClusterState oldState, ClusterS
13071316
* name. If the repo uuid was not known at the time the snapshot was added to {@link SnapshotDeletionsPending}, we try to find a
13081317
* repository with the same name.
13091318
*
1310-
* @param state the current {@link ClusterState}
1319+
* @param event the current {@link ClusterChangedEvent}
13111320
*/
1312-
private void triggerSnapshotsPendingDeletions(final ClusterState state) {
1313-
final RepositoriesMetadata repositories = state.metadata().custom(RepositoriesMetadata.TYPE, RepositoriesMetadata.EMPTY);
1314-
final SnapshotDeletionsPending snapshotDeletionsPending = state.custom(SnapshotDeletionsPending.TYPE);
1315-
if (snapshotDeletionsPending == null
1316-
|| snapshotDeletionsPending.isEmpty()
1317-
|| repositories.repositories().isEmpty()
1318-
|| state.nodes().isLocalNodeElectedMaster() == false
1319-
|| state.custom(RepositoryCleanupInProgress.TYPE, RepositoryCleanupInProgress.EMPTY).hasCleanupInProgress()) {
1321+
private void triggerSnapshotsPendingDeletions(final ClusterChangedEvent event) {
1322+
if (event.localNodeMaster() == false) {
1323+
if (event.previousState().nodes().isLocalNodeElectedMaster()) {
1324+
clearPendingDeletionsWithConflicts();
1325+
}
13201326
return;
13211327
}
1322-
final Set<SnapshotId> currentDeletions = deletionsSources(state);
1323-
final Set<SnapshotId> currentRestores = restoreSources(state);
1324-
final Set<SnapshotId> currentClones = cloneSources(state);
13251328

1326-
// the list of snapshot ids to trigger deletion for, per repository
1327-
final Map<RepositoryMetadata, Set<SnapshotId>> snapshotsToDelete = new HashMap<>();
1329+
if (pendingDeletionsChanged(event) || pendingDeletionsWithConflictsChanged(event)) {
1330+
final ClusterState state = event.state();
1331+
final SnapshotDeletionsPending snapshotDeletionsPending = state.custom(SnapshotDeletionsPending.TYPE);
1332+
if (snapshotDeletionsPending == null || snapshotDeletionsPending.isEmpty()) {
1333+
clearPendingDeletionsWithConflicts();
1334+
return;
1335+
}
13281336

1329-
for (SnapshotDeletionsPending.Entry snapshot : snapshotDeletionsPending.entries()) {
1330-
final SnapshotId snapshotId = snapshot.getSnapshotId();
1337+
final RepositoriesMetadata repositories = state.metadata().custom(RepositoriesMetadata.TYPE, RepositoriesMetadata.EMPTY);
13311338

1332-
if (currentRestores.contains(snapshotId)) {
1333-
logger.trace("snapshot to delete [{}] is being restored, waiting for restore to complete", snapshotId);
1334-
continue;
1335-
} else if (currentClones.contains(snapshotId)) {
1336-
logger.trace("snapshot to delete [{}] is being cloned, waiting for cloning to complete", snapshotId);
1337-
continue;
1338-
} else if (currentDeletions.contains(snapshotId)) {
1339-
logger.trace("snapshot to delete [{}] is already queued", snapshotId);
1340-
continue;
1341-
}
1339+
final Set<SnapshotId> currentDeletions = deletionsSources(state);
1340+
final Set<SnapshotId> currentRestores = restoreSources(state);
1341+
final Set<SnapshotId> currentClones = cloneSources(state);
13421342

1343-
final Optional<RepositoryMetadata> optionalRepository = findRepositoryForPendingDeletion(
1344-
repositories,
1345-
snapshot.getRepositoryName(),
1346-
snapshot.getRepositoryUuid()
1347-
);
1348-
if (optionalRepository.isEmpty()) {
1349-
logger.debug(
1350-
"repository [{}/{}] not found, cannot delete pending snapshot [{}] created at {}",
1343+
// the list of snapshot ids to trigger deletion for, per repository
1344+
final Map<RepositoryMetadata, Set<SnapshotId>> snapshotsToDelete = new HashMap<>();
1345+
1346+
for (SnapshotDeletionsPending.Entry snapshot : snapshotDeletionsPending.entries()) {
1347+
final SnapshotId snapshotId = snapshot.getSnapshotId();
1348+
1349+
if (currentRestores.contains(snapshotId)) {
1350+
logger.trace("snapshot to delete [{}] is being restored, waiting for restore to complete", snapshotId);
1351+
pendingDeletionsWithConflictingRestores.add(snapshotId);
1352+
continue;
1353+
}
1354+
pendingDeletionsWithConflictingRestores.remove(snapshotId);
1355+
1356+
if (currentClones.contains(snapshotId)) {
1357+
logger.trace("snapshot to delete [{}] is being cloned, waiting for cloning to complete", snapshotId);
1358+
pendingDeletionsWithConflictingClones.add(snapshotId);
1359+
continue;
1360+
}
1361+
pendingDeletionsWithConflictingClones.remove(snapshotId);
1362+
1363+
if (currentDeletions.contains(snapshotId)) {
1364+
logger.trace("snapshot to delete [{}] is already queued", snapshotId);
1365+
pendingDeletionsWithConflictingRepos.remove(snapshotId);
1366+
continue;
1367+
}
1368+
1369+
if (state.custom(RepositoryCleanupInProgress.TYPE, RepositoryCleanupInProgress.EMPTY).hasCleanupInProgress()) {
1370+
if (pendingDeletionsWithConflictingRepos.add(snapshotId)) {
1371+
logger.debug(
1372+
"a repository clean-up is in progress, cannot delete pending snapshot [{}] created at {}",
1373+
snapshotId,
1374+
Instant.ofEpochMilli(snapshot.getIndexDeletionTime()).atZone(ZoneOffset.UTC)
1375+
);
1376+
}
1377+
continue;
1378+
}
1379+
1380+
final Optional<RepositoryMetadata> optionalRepository = findRepositoryForPendingDeletion(
1381+
repositories,
13511382
snapshot.getRepositoryName(),
1352-
snapshot.getRepositoryUuid(),
1353-
snapshotId,
1354-
Instant.ofEpochMilli(snapshot.getIndexDeletionTime()).atZone(ZoneOffset.UTC)
1383+
snapshot.getRepositoryUuid()
13551384
);
1356-
continue;
1385+
if (optionalRepository.isEmpty()) {
1386+
if (pendingDeletionsWithConflictingRepos.add(snapshotId)) {
1387+
logger.debug(
1388+
"repository [{}/{}] not found, cannot delete pending snapshot [{}] created at {}",
1389+
snapshot.getRepositoryName(),
1390+
snapshot.getRepositoryUuid(),
1391+
snapshotId,
1392+
Instant.ofEpochMilli(snapshot.getIndexDeletionTime()).atZone(ZoneOffset.UTC)
1393+
);
1394+
}
1395+
continue;
1396+
}
1397+
final RepositoryMetadata repository = optionalRepository.get();
1398+
if (repository.settings().getAsBoolean(READONLY_SETTING_KEY, false)) {
1399+
if (pendingDeletionsWithConflictingRepos.add(snapshotId)) {
1400+
logger.debug(
1401+
"repository [{}/{}] is read-only, cannot delete pending snapshot [{}] created at {}",
1402+
repository.name(),
1403+
repository.uuid(),
1404+
snapshotId,
1405+
Instant.ofEpochMilli(snapshot.getIndexDeletionTime()).atZone(ZoneOffset.UTC)
1406+
);
1407+
}
1408+
continue;
1409+
}
1410+
pendingDeletionsWithConflictingRepos.remove(snapshotId);
1411+
1412+
// should we add some throttling to not always retry?
1413+
if (ongoingSnapshotsDeletions.add(snapshotId)) {
1414+
logger.info("triggering snapshot deletion for [{}]", snapshotId);
1415+
final boolean added = snapshotsToDelete.computeIfAbsent(repository, r -> new HashSet<>()).add(snapshotId);
1416+
assert added : snapshotId;
1417+
}
13571418
}
1419+
snapshotsToDelete.forEach(
1420+
(repo, snapshots) -> threadPool.generic().execute(new SnapshotsToDeleteRunnable(repo.name(), repo.uuid(), snapshots))
1421+
);
1422+
}
1423+
}
13581424

1359-
final RepositoryMetadata repository = optionalRepository.get();
1360-
if (repository.settings().getAsBoolean(READONLY_SETTING_KEY, false)) {
1361-
logger.debug(
1362-
"repository [{}/{}] is read-only, cannot delete pending snapshot [{}] created at {}",
1363-
repository.name(),
1364-
repository.uuid(),
1365-
snapshotId,
1366-
Instant.ofEpochMilli(snapshot.getIndexDeletionTime()).atZone(ZoneOffset.UTC)
1367-
);
1368-
continue;
1425+
private void clearPendingDeletionsWithConflicts() {
1426+
pendingDeletionsWithConflictingRestores.clear();
1427+
pendingDeletionsWithConflictingClones.clear();
1428+
pendingDeletionsWithConflictingRepos.clear();
1429+
}
1430+
1431+
private static boolean pendingDeletionsChanged(ClusterChangedEvent event) {
1432+
SnapshotDeletionsPending previous = event.previousState().custom(SnapshotDeletionsPending.TYPE, SnapshotDeletionsPending.EMPTY);
1433+
SnapshotDeletionsPending currents = event.state().custom(SnapshotDeletionsPending.TYPE, SnapshotDeletionsPending.EMPTY);
1434+
return Objects.equals(previous, currents) == false;
1435+
}
1436+
1437+
private boolean pendingDeletionsWithConflictsChanged(ClusterChangedEvent event) {
1438+
if (pendingDeletionsWithConflictingRestores.isEmpty() == false) {
1439+
RestoreInProgress previous = event.previousState().custom(RestoreInProgress.TYPE, RestoreInProgress.EMPTY);
1440+
RestoreInProgress currents = event.state().custom(RestoreInProgress.TYPE, RestoreInProgress.EMPTY);
1441+
if (Objects.equals(previous, currents) == false) {
1442+
return true;
1443+
}
1444+
}
1445+
if (pendingDeletionsWithConflictingClones.isEmpty() == false) {
1446+
Set<SnapshotsInProgress.Entry> previous = event.previousState()
1447+
.custom(SnapshotsInProgress.TYPE, SnapshotsInProgress.EMPTY)
1448+
.asStream()
1449+
.filter(SnapshotsInProgress.Entry::isClone)
1450+
.collect(Collectors.toSet());
1451+
Set<SnapshotsInProgress.Entry> currents = event.state()
1452+
.custom(SnapshotsInProgress.TYPE, SnapshotsInProgress.EMPTY)
1453+
.asStream()
1454+
.filter(SnapshotsInProgress.Entry::isClone)
1455+
.collect(Collectors.toSet());
1456+
if (Objects.equals(previous, currents) == false) {
1457+
return true;
1458+
}
1459+
}
1460+
if (pendingDeletionsWithConflictingRepos.isEmpty() == false) {
1461+
boolean previousCleanUp = event.previousState()
1462+
.custom(RepositoryCleanupInProgress.TYPE, RepositoryCleanupInProgress.EMPTY)
1463+
.hasCleanupInProgress();
1464+
boolean currentCleanUp = event.state()
1465+
.custom(RepositoryCleanupInProgress.TYPE, RepositoryCleanupInProgress.EMPTY)
1466+
.hasCleanupInProgress();
1467+
if (previousCleanUp != currentCleanUp) {
1468+
return true;
13691469
}
13701470

1371-
// should we add some throttling to not always retry?
1372-
if (ongoingSnapshotsDeletions.add(snapshotId)) {
1373-
logger.trace("triggering snapshot deletion for [{}]", snapshotId);
1374-
final boolean added = snapshotsToDelete.computeIfAbsent(repository, r -> new HashSet<>()).add(snapshotId);
1375-
assert added : snapshotId;
1471+
RepositoriesMetadata previous = event.previousState().metadata().custom(RepositoriesMetadata.TYPE, RepositoriesMetadata.EMPTY);
1472+
RepositoriesMetadata current = event.state().metadata().custom(RepositoriesMetadata.TYPE, RepositoriesMetadata.EMPTY);
1473+
if (previous.equals(current) == false) {
1474+
return true;
13761475
}
13771476
}
1378-
snapshotsToDelete.forEach(
1379-
(repo, snapshots) -> threadPool.generic().execute(new SnapshotsToDeleteRunnable(repo.name(), repo.uuid(), snapshots))
1380-
);
1477+
return false;
13811478
}
13821479

13831480
/**
@@ -1426,6 +1523,7 @@ public void onFailure(Exception e) {
14261523
shouldRetry = RepositoryData.MISSING_UUID.equals(repositoryUuid) == false;
14271524

14281525
} else if (e instanceof ConcurrentSnapshotExecutionException) {
1526+
assert false : e;
14291527
logger.debug(
14301528
"[{}] failed to delete snapshot [{}]: a concurrent operation is running",
14311529
repositoryName,

‎x-pack/plugin/searchable-snapshots/src/internalClusterTest/java/org/elasticsearch/xpack/searchablesnapshots/SearchableSnapshotsPendingDeletionsIntegTests.java

Lines changed: 1 addition & 1 deletion
Original file line numberDiff line numberDiff line change
@@ -94,7 +94,7 @@ public void testSearchableSnapshotIsDeletedWhenRepoIsRecreated() throws Exceptio
9494
} else {
9595
// re register the repository under a different name: the snapshot
9696
// pending deletion logic should try to delete the snapshot based
97-
// on the repository uuid
97+
// on the repository uuid, that is why we force a verification here
9898
repoName = "new_" + repository;
9999
createRepository(repoName, "mock", repositorySettings, true);
100100
}

0 commit comments

Comments
 (0)