Skip to content
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

[WIP] Keep SnapshotInProgress State in Sync with Routing Table #35637

Closed
wants to merge 1 commit into from
Closed

[WIP] Keep SnapshotInProgress State in Sync with Routing Table #35637

Show file tree
Hide file tree
Changes from all commits
Commits
Show all changes
1 commit
Select commit
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
19 changes: 17 additions & 2 deletions server/src/main/java/org/elasticsearch/cluster/routing/allocation/AllocationService.java
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -23,6 +23,7 @@
import org.elasticsearch.cluster.ClusterInfoService;
import org.elasticsearch.cluster.ClusterState;
import org.elasticsearch.cluster.RestoreInProgress;
import org.elasticsearch.cluster.SnapshotsInProgress;
import org.elasticsearch.cluster.health.ClusterHealthStatus;
import org.elasticsearch.cluster.health.ClusterStateHealth;
import org.elasticsearch.cluster.metadata.AutoExpandReplicas;
Expand Down Expand Up @@ -134,14 +135,28 @@ private ClusterState buildResult(ClusterState oldState, RoutingAllocation alloca
.routingTable(newRoutingTable)
.metaData(newMetaData);
final RestoreInProgress restoreInProgress = allocation.custom(RestoreInProgress.TYPE);
ImmutableOpenMap.Builder<String, ClusterState.Custom> customsBuilder = null;
if (restoreInProgress != null) {
RestoreInProgress updatedRestoreInProgress = allocation.updateRestoreInfoWithRoutingChanges(restoreInProgress);
if (updatedRestoreInProgress != restoreInProgress) {
ImmutableOpenMap.Builder<String, ClusterState.Custom> customsBuilder = ImmutableOpenMap.builder(allocation.getCustoms());
customsBuilder = ImmutableOpenMap.builder(allocation.getCustoms());
customsBuilder.put(RestoreInProgress.TYPE, updatedRestoreInProgress);
newStateBuilder.customs(customsBuilder.build());
}
}
final SnapshotsInProgress snapshotsInProgress = allocation.custom(SnapshotsInProgress.TYPE);
if (snapshotsInProgress != null) {
SnapshotsInProgress updatedSnapshotsInProgress =
allocation.updateSnapshotsWithRoutingChanges(snapshotsInProgress, newRoutingTable);
if (updatedSnapshotsInProgress != snapshotsInProgress) {
if (customsBuilder == null) {
customsBuilder = ImmutableOpenMap.builder(allocation.getCustoms());
}
customsBuilder.put(SnapshotsInProgress.TYPE, updatedSnapshotsInProgress);
}
}
if (customsBuilder != null) {
newStateBuilder.customs(customsBuilder.build());
}
return newStateBuilder.build();
}

Expand Down
15 changes: 13 additions & 2 deletions server/src/main/java/org/elasticsearch/cluster/routing/allocation/RoutingAllocation.java
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -22,6 +22,7 @@
import org.elasticsearch.cluster.ClusterInfo;
import org.elasticsearch.cluster.ClusterState;
import org.elasticsearch.cluster.RestoreInProgress;
import org.elasticsearch.cluster.SnapshotsInProgress;
import org.elasticsearch.cluster.metadata.MetaData;
import org.elasticsearch.cluster.node.DiscoveryNodes;
import org.elasticsearch.cluster.routing.RoutingChangesObserver;
Expand All @@ -38,6 +39,7 @@
import java.util.HashSet;
import java.util.Map;
import java.util.Set;
import org.elasticsearch.snapshots.SnapshotsService;

import static java.util.Collections.emptySet;
import static java.util.Collections.unmodifiableSet;
Expand Down Expand Up @@ -76,11 +78,12 @@ public class RoutingAllocation {
private final IndexMetaDataUpdater indexMetaDataUpdater = new IndexMetaDataUpdater();
private final RoutingNodesChangedObserver nodesChangedObserver = new RoutingNodesChangedObserver();
private final RestoreInProgressUpdater restoreInProgressUpdater = new RestoreInProgressUpdater();
private final SnapshotsService.SnapshotsInProgressUpdater snapshotsInProgressUpdater =
new SnapshotsService.SnapshotsInProgressUpdater();
private final RoutingChangesObserver routingChangesObserver = new RoutingChangesObserver.DelegatingRoutingChangesObserver(
nodesChangedObserver, indexMetaDataUpdater, restoreInProgressUpdater
nodesChangedObserver, indexMetaDataUpdater, restoreInProgressUpdater, snapshotsInProgressUpdater
);


/**
* Creates a new {@link RoutingAllocation}
* @param deciders {@link AllocationDeciders} to used to make decisions for routing allocations
Expand Down Expand Up @@ -251,6 +254,14 @@ public RestoreInProgress updateRestoreInfoWithRoutingChanges(RestoreInProgress r
return restoreInProgressUpdater.applyChanges(restoreInProgress);
}

/**
* Returns updated {@link SnapshotsInProgress} based on the changes that were made to the routing nodes
*/
public SnapshotsInProgress updateSnapshotsWithRoutingChanges(SnapshotsInProgress snapshotsInProgress,
RoutingTable newRoutingTable) {
return snapshotsInProgressUpdater.applyChanges(snapshotsInProgress, newRoutingTable);
}

/**
* Returns true iff changes were made to the routing nodes
*/
Expand Down
156 changes: 156 additions & 0 deletions server/src/main/java/org/elasticsearch/snapshots/SnapshotsService.java
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -43,8 +43,10 @@
import org.elasticsearch.cluster.node.DiscoveryNodes;
import org.elasticsearch.cluster.routing.IndexRoutingTable;
import org.elasticsearch.cluster.routing.IndexShardRoutingTable;
import org.elasticsearch.cluster.routing.RoutingChangesObserver;
import org.elasticsearch.cluster.routing.RoutingTable;
import org.elasticsearch.cluster.routing.ShardRouting;
import org.elasticsearch.cluster.routing.UnassignedInfo;
import org.elasticsearch.cluster.service.ClusterService;
import org.elasticsearch.common.Nullable;
import org.elasticsearch.common.Priority;
Expand Down Expand Up @@ -1536,6 +1538,160 @@ public interface SnapshotCompletionListener {
void onSnapshotFailure(Snapshot snapshot, Exception e);
}

public static final class SnapshotsInProgressUpdater extends RoutingChangesObserver.AbstractRoutingChangesObserver {

private final Set<ShardId> shardChanges = new HashSet<>();

public SnapshotsInProgress applyChanges(SnapshotsInProgress oldSnapshot, RoutingTable newRoutingTable) {
return updateWithRoutingTable(shardChanges, oldSnapshot, newRoutingTable);
}

@Override
public void shardInitialized(ShardRouting unassignedShard, ShardRouting initializedShard) {
onChanged(unassignedShard.shardId());
}

@Override
public void shardStarted(ShardRouting initializingShard, ShardRouting startedShard) {
onChanged(initializingShard.shardId());
}
@Override
public void relocationStarted(ShardRouting startedShard, ShardRouting targetRelocatingShard) {
onChanged(startedShard.shardId());
}
@Override
public void unassignedInfoUpdated(ShardRouting unassignedShard, UnassignedInfo newUnassignedInfo) {
onChanged(unassignedShard.shardId());
}
@Override
public void shardFailed(ShardRouting failedShard, UnassignedInfo unassignedInfo) {
onChanged(failedShard.shardId());
}
@Override
public void relocationCompleted(ShardRouting removedRelocationSource) {
onChanged(removedRelocationSource.shardId());
}
@Override
public void relocationSourceRemoved(ShardRouting removedReplicaRelocationSource) {
onChanged(removedReplicaRelocationSource.shardId());
}
@Override
public void startedPrimaryReinitialized(ShardRouting startedPrimaryShard, ShardRouting initializedShard) {
onChanged(startedPrimaryShard.shardId());
}
@Override
public void replicaPromoted(ShardRouting replicaShard) {
onChanged(replicaShard.shardId());
}
@Override
public void initializedReplicaReinitialized(ShardRouting oldReplica, ShardRouting reinitializedReplica) {
onChanged(oldReplica.shardId());
}

private void onChanged(ShardId shardId) {
shardChanges.add(shardId);
}

private static SnapshotsInProgress updateWithRoutingTable(Set<ShardId> shardIds, SnapshotsInProgress oldSnapshot,
RoutingTable newRoutingTable) {
if (oldSnapshot == null || shardIds.isEmpty()) {
return oldSnapshot;
}
List<SnapshotsInProgress.Entry> entries = new ArrayList<>();
boolean snapshotsInProgressChanged = false;
for (SnapshotsInProgress.Entry entry : oldSnapshot.entries()) {
ImmutableOpenMap.Builder<ShardId, ShardSnapshotStatus> shardsBuilder = null;
for (ShardId shardId : shardIds) {
ImmutableOpenMap<ShardId, ShardSnapshotStatus> shards = entry.shards();
ShardSnapshotStatus currentStatus = shards.get(shardId);
if (currentStatus != null) {
final State currentState = currentStatus.state();
if (currentState.completed() == false) {
ShardRouting primaryShardRouting = Optional
.ofNullable(newRoutingTable.shardRoutingTableOrNull(shardId))
.map(IndexShardRoutingTable::primaryShard)
.orElse(null);
final ShardSnapshotStatus newStatus;
if (primaryShardRouting == null) {
newStatus = failedStatus(null, "missing shard");
} else if (primaryShardRouting.active() == false) {
if (primaryShardRouting.initializing() && currentState == State.WAITING) {
newStatus = currentStatus;
} else {
newStatus = failedStatus(
primaryShardRouting.currentNodeId(),
primaryShardRouting.unassignedInfo().getReason().toString()
);
}
} else if (primaryShardRouting.started()) {
switch (currentState) {
case WAITING:
newStatus = new ShardSnapshotStatus(primaryShardRouting.currentNodeId());
break;
case INIT: {
String currentNodeId = currentStatus.nodeId();
if (primaryShardRouting.currentNodeId().equals(currentNodeId)) {
newStatus = currentStatus;
} else {
newStatus = failedStatus(currentNodeId);
}
break;
}
case ABORTED:
String currentNodeId = currentStatus.nodeId();
if (currentNodeId.equals(primaryShardRouting.currentNodeId())) {
newStatus = currentStatus;
} else {
newStatus = failedStatus(currentNodeId);
}
break;
default:
newStatus = currentStatus;
break;
}
} else if (currentState == State.INIT || currentStatus.state() == State.ABORTED) {
newStatus = failedStatus(currentStatus.nodeId());
} else {
newStatus = currentStatus;
}
if (newStatus != currentStatus) {
if (shardsBuilder == null) {
shardsBuilder = ImmutableOpenMap.builder(shards);
}
shardsBuilder.put(shardId, newStatus);
}
}
}
}
if (shardsBuilder == null) {
entries.add(entry);
} else {
snapshotsInProgressChanged = true;
ImmutableOpenMap<ShardId, ShardSnapshotStatus> shards = shardsBuilder.build();
entries.add(
new SnapshotsInProgress.Entry(
entry,
completed(shards.values()) ? State.SUCCESS : entry.state(),
shards
)
);
}
}
if (snapshotsInProgressChanged) {
return new SnapshotsInProgress(entries);
}
return oldSnapshot;
}

private static ShardSnapshotStatus failedStatus(String nodeId) {
return failedStatus(nodeId, "shard failed");
}

private static ShardSnapshotStatus failedStatus(String nodeId, String reason) {
return new ShardSnapshotStatus(nodeId, State.FAILED, reason);
}
}

/**
* Snapshot creation request
*/
Expand Down