Skip to content

Commit

Permalink
Avoid Needless Cache Status Fetches in SearchableSnapshotAllocator (#…
Browse files Browse the repository at this point in the history 
…66433)

We shouldn't fetch cache status if no allocation is possible to begin with.
Also, this surfaced an issue with using the `Client` to `reroute` since that
won't retry stale shards (failed the invalid license IT for example) so I moved
to using the `RerouteService` like we do in the `GatewayAllocator`.
(Plus, dried up one method that was 100% the same as in the replica allocator)
  • Loading branch information
@original-brownbear
original-brownbear authored Dec 16, 2020
1 parent 6fe1921 commit 7c65a2b
Show file tree
Hide file tree
Showing 5 changed files with 133 additions and 109 deletions.
2 changes: 1 addition & 1 deletion .../src/main/java/org/elasticsearch/action/admin/cluster/reroute/ClusterRerouteResponse.java
View file
Original file line number Diff line number Diff line change
Expand Up @@ -44,7 +44,7 @@ public class ClusterRerouteResponse extends AcknowledgedResponse implements ToXC
explanations = RoutingExplanations.readFrom(in);
}

public ClusterRerouteResponse(boolean acknowledged, ClusterState state, RoutingExplanations explanations) {
ClusterRerouteResponse(boolean acknowledged, ClusterState state, RoutingExplanations explanations) {
super(acknowledged);
this.state = state;
this.explanations = explanations;
Expand Down
4 changes: 2 additions & 2 deletions server/src/main/java/org/elasticsearch/gateway/ReplicaShardAllocator.java
View file
Original file line number Diff line number Diff line change
Expand Up @@ -241,8 +241,8 @@ public AllocateUnassignedDecision makeAllocationDecision(final ShardRouting unas
* YES or THROTTLE). If in explain mode, also returns the node-level explanations as the second element
* in the returned tuple.
*/
private static Tuple<Decision, Map<String, NodeAllocationResult>> canBeAllocatedToAtLeastOneNode(ShardRouting shard,
RoutingAllocation allocation) {
public static Tuple<Decision, Map<String, NodeAllocationResult>> canBeAllocatedToAtLeastOneNode(ShardRouting shard,
RoutingAllocation allocation) {
Decision madeDecision = Decision.NO;
final boolean explain = allocation.debugDecision();
Map<String, NodeAllocationResult> nodeDecisions = explain ? new HashMap<>() : null;
Expand Down
77 changes: 23 additions & 54 deletions ...rc/main/java/org/elasticsearch/xpack/searchablesnapshots/SearchableSnapshotAllocator.java
View file
Original file line number Diff line number Diff line change
Expand Up @@ -5,17 +5,17 @@
*/
package org.elasticsearch.xpack.searchablesnapshots;

import com.carrotsearch.hppc.cursors.ObjectCursor;
import org.apache.logging.log4j.LogManager;
import org.apache.logging.log4j.Logger;
import org.elasticsearch.Version;
import org.elasticsearch.action.ActionListener;
import org.elasticsearch.action.FailedNodeException;
import org.elasticsearch.action.admin.cluster.reroute.ClusterRerouteResponse;
import org.elasticsearch.client.Client;
import org.elasticsearch.cluster.ClusterState;
import org.elasticsearch.cluster.node.DiscoveryNode;
import org.elasticsearch.cluster.node.DiscoveryNodes;
import org.elasticsearch.cluster.routing.RecoverySource;
import org.elasticsearch.cluster.routing.RerouteService;
import org.elasticsearch.cluster.routing.RoutingNode;
import org.elasticsearch.cluster.routing.ShardRouting;
import org.elasticsearch.cluster.routing.UnassignedInfo;
Expand All @@ -27,11 +27,13 @@
import org.elasticsearch.cluster.routing.allocation.RoutingAllocation;
import org.elasticsearch.cluster.routing.allocation.decider.Decision;
import org.elasticsearch.common.Nullable;
import org.elasticsearch.common.Priority;
import org.elasticsearch.common.collect.Tuple;
import org.elasticsearch.common.settings.Settings;
import org.elasticsearch.common.unit.ByteSizeValue;
import org.elasticsearch.common.util.concurrent.ConcurrentCollections;
import org.elasticsearch.gateway.AsyncShardFetch;
import org.elasticsearch.gateway.ReplicaShardAllocator;
import org.elasticsearch.index.shard.ShardId;
import org.elasticsearch.repositories.IndexId;
import org.elasticsearch.snapshots.Snapshot;
Expand Down Expand Up @@ -60,9 +62,9 @@ public class SearchableSnapshotAllocator implements ExistingShardsAllocator {

private static final Logger logger = LogManager.getLogger(SearchableSnapshotAllocator.class);

private static final ActionListener<ClusterRerouteResponse> REROUTE_LISTENER = new ActionListener<>() {
private static final ActionListener<ClusterState> REROUTE_LISTENER = new ActionListener<>() {
@Override
public void onResponse(ClusterRerouteResponse clusterRerouteResponse) {
public void onResponse(ClusterState clusterRerouteResponse) {
logger.trace("reroute succeeded after loading snapshot cache information");
}

Expand All @@ -78,8 +80,11 @@ public void onFailure(Exception e) {

private final Client client;

public SearchableSnapshotAllocator(Client client) {
private final RerouteService rerouteService;

public SearchableSnapshotAllocator(Client client, RerouteService rerouteService) {
this.client = client;
this.rerouteService = rerouteService;
}

@Override
Expand Down Expand Up @@ -151,20 +156,15 @@ private AllocateUnassignedDecision decideAllocation(RoutingAllocation allocation
return AllocateUnassignedDecision.no(UnassignedInfo.AllocationStatus.FETCHING_SHARD_DATA, null);
}

final AsyncShardFetch.FetchResult<NodeCacheFilesMetadata> fetchedCacheData = fetchData(shardRouting, allocation);
if (fetchedCacheData.hasData() == false) {
return AllocateUnassignedDecision.no(UnassignedInfo.AllocationStatus.FETCHING_SHARD_DATA, null);
}

final boolean explain = allocation.debugDecision();
final MatchingNodes matchingNodes = findMatchingNodes(shardRouting, allocation, fetchedCacheData, explain);
assert explain == false || matchingNodes.nodeDecisions != null : "in explain mode, we must have individual node decisions";

// pre-check if it can be allocated to any node that currently exists, so we won't list the cache sizes for it for nothing
// TODO: in the following logic, we do not account for existing cache size when handling disk space checks, should and can we
// reliably do this in a world of concurrent cache evictions or are we ok with the cache size just being a best effort hint
// here?
Tuple<Decision, Map<String, NodeAllocationResult>> result = canBeAllocatedToAtLeastOneNode(shardRouting, allocation);
Tuple<Decision, Map<String, NodeAllocationResult>> result = ReplicaShardAllocator.canBeAllocatedToAtLeastOneNode(
shardRouting,
allocation
);
Decision allocateDecision = result.v1();
if (allocateDecision.type() != Decision.Type.YES && (explain == false || asyncFetchStore.get(shardRouting.shardId()) == null)) {
// only return early if we are not in explain mode, or we are in explain mode but we have not
Expand All @@ -176,6 +176,14 @@ private AllocateUnassignedDecision decideAllocation(RoutingAllocation allocation
);
}

final AsyncShardFetch.FetchResult<NodeCacheFilesMetadata> fetchedCacheData = fetchData(shardRouting, allocation);
if (fetchedCacheData.hasData() == false) {
return AllocateUnassignedDecision.no(UnassignedInfo.AllocationStatus.FETCHING_SHARD_DATA, null);
}

final MatchingNodes matchingNodes = findMatchingNodes(shardRouting, allocation, fetchedCacheData, explain);
assert explain == false || matchingNodes.nodeDecisions != null : "in explain mode, we must have individual node decisions";

List<NodeAllocationResult> nodeDecisions = augmentExplanationsWithStoreInfo(result.v2(), matchingNodes.nodeDecisions);
if (allocateDecision.type() != Decision.Type.YES) {
return AllocateUnassignedDecision.no(UnassignedInfo.AllocationStatus.fromDecision(allocateDecision.type()), nodeDecisions);
Expand Down Expand Up @@ -283,7 +291,7 @@ public void onFailure(Exception e) {
}
}, () -> {
if (asyncFetch.data() != null) {
client.admin().cluster().prepareReroute().execute(REROUTE_LISTENER);
rerouteService.reroute("async_shard_cache_fetch", Priority.HIGH, REROUTE_LISTENER);
}
})
);
Expand Down Expand Up @@ -313,45 +321,6 @@ private static List<NodeAllocationResult> augmentExplanationsWithStoreInfo(
return augmented;
}

/**
* Determines if the shard can be allocated on at least one node based on the allocation deciders.
*
* Returns the best allocation decision for allocating the shard on any node (i.e. YES if at least one
* node decided YES, THROTTLE if at least one node decided THROTTLE, and NO if none of the nodes decided
* YES or THROTTLE). If in explain mode, also returns the node-level explanations as the second element
* in the returned tuple.
* TODO: dry this method up against ReplicaShardAllocator
*/
private static Tuple<Decision, Map<String, NodeAllocationResult>> canBeAllocatedToAtLeastOneNode(
ShardRouting shard,
RoutingAllocation allocation
) {
Decision madeDecision = Decision.NO;
final boolean explain = allocation.debugDecision();
Map<String, NodeAllocationResult> nodeDecisions = explain ? new HashMap<>() : null;
for (ObjectCursor<DiscoveryNode> cursor : allocation.nodes().getDataNodes().values()) {
RoutingNode node = allocation.routingNodes().node(cursor.value.getId());
if (node == null) {
continue;
}
// if we can't allocate it on a node, ignore it
Decision decision = allocation.deciders().canAllocate(shard, node, allocation);
if (decision.type() == Decision.Type.YES && madeDecision.type() != Decision.Type.YES) {
if (explain) {
madeDecision = decision;
} else {
return Tuple.tuple(decision, null);
}
} else if (madeDecision.type() == Decision.Type.NO && decision.type() == Decision.Type.THROTTLE) {
madeDecision = decision;
}
if (explain) {
nodeDecisions.put(node.nodeId(), new NodeAllocationResult(node.node(), null, decision));
}
}
return Tuple.tuple(madeDecision, nodeDecisions);
}

private MatchingNodes findMatchingNodes(
ShardRouting shard,
RoutingAllocation allocation,
Expand Down
6 changes: 3 additions & 3 deletions ...pshots/src/main/java/org/elasticsearch/xpack/searchablesnapshots/SearchableSnapshots.java
View file
Original file line number Diff line number Diff line change
Expand Up @@ -172,8 +172,8 @@ public class SearchableSnapshots extends Plugin implements IndexStorePlugin, Eng
private final SetOnce<BlobStoreCacheService> blobStoreCacheService = new SetOnce<>();
private final SetOnce<CacheService> cacheService = new SetOnce<>();
private final SetOnce<ThreadPool> threadPool = new SetOnce<>();
private final SetOnce<Client> client = new SetOnce<>();
private final SetOnce<FailShardsOnInvalidLicenseClusterListener> failShardsListener = new SetOnce<>();
private final SetOnce<SearchableSnapshotAllocator> allocator = new SetOnce<>();
private final Settings settings;

public SearchableSnapshots(final Settings settings) {
Expand Down Expand Up @@ -235,7 +235,7 @@ public Collection<Object> createComponents(
} else {
PersistentCache.cleanUp(settings, nodeEnvironment);
}
this.client.set(client);
this.allocator.set(new SearchableSnapshotAllocator(client, clusterService.getRerouteService()));
components.add(new CacheServiceSupplier(cacheService.get()));
return Collections.unmodifiableList(components);
}
Expand Down Expand Up @@ -344,7 +344,7 @@ public List<RestHandler> getRestHandlers(

@Override
public Map<String, ExistingShardsAllocator> getExistingShardsAllocators() {
return Map.of(SearchableSnapshotAllocator.ALLOCATOR_NAME, new SearchableSnapshotAllocator(client.get()));
return Map.of(SearchableSnapshotAllocator.ALLOCATOR_NAME, allocator.get());
}

// overridable by tests
Expand Down
Loading

0 comments on commit 7c65a2b

Please sign in to comment.