Resharding Add Flush and Refresh functionality #136880
Conversation
elasticsearchmachine
added
v9.3.0
serverless-linked
…ticsearch into 10202025/ReshardFlushSplit pull
elasticsearchmachine
added
the
needs:triage
ankikuma
added
:Distributed Indexing/Distributed
elasticsearchmachine
removed
the
needs:triage
|
Pinging @elastic/es-distributed-indexing (Team:Distributed Indexing) |
elasticsearchmachine
removed
the
needs:triage
|
|
||
| // We are here because there was a mismatch between the SplitShardCountSummary in the request | ||
| // and that on the primary shard node. We assume that the request is exactly 1 reshard split behind | ||
| // the current state. |
There was a problem hiding this comment.
This assumption will need to be revised in a follow up PR. I created ES-13413 for this.
| // If the action fails on either one of the shards, we return an exception. | ||
| // Case 1: Both source and target shards return a response: Add up total, successful, failures | ||
| // Case 2: Both source and target shards return an exception : return exception | ||
| // Case 3: One shards returns a response, the other returns an exception : return exception |
There was a problem hiding this comment.
We fail the entire request with an exception if the action fails on either one of the 2 shards. Please comment if you think this behavior is incorrect.
| assert failureArray.length == failed; | ||
| ReplicationResponse.ShardInfo shardInfo = ReplicationResponse.ShardInfo.of(total, successful, failureArray); | ||
| ReplicationResponse response = new ReplicationResponse(); | ||
| response.setShardInfo(shardInfo); |
There was a problem hiding this comment.
ShardInfo adds up the successes and failures from the 2 responses.
| Map<ShardId, ShardFlushRequest> requestsByShard = new HashMap<>(); | ||
| requestsByShard.put(sourceShard, request); | ||
| // Create a request for original source shard and for each target shard. | ||
| // New requests that are to be handled by target shards should contain the |
There was a problem hiding this comment.
We should be careful with using cluster state on the source shard node. During handoff, the target shard node applies a non-acked cluster state update to transition to HANDOFF and then acks handoff request from the source shard. At this point there is no guarantee that this update to HANDOFF was observed and that SplitShardCountSummary.forIndexing returns a correct result. In practice it shouldn't really matter though because a particular request can only be split once?
| // latest ShardCountSummary. | ||
| int targetShardId = indexMetadata.getReshardingMetadata().getSplit().targetShard(sourceShard.id()); | ||
| ShardId targetShard = new ShardId(request.shardId().getIndex(), targetShardId); | ||
| requestsByShard.put(targetShard, new ShardFlushRequest(request.getRequest(), targetShard, shardCountSummary)); |
There was a problem hiding this comment.
Do we want to skip this logic if shardCountSummary is up to date (the caveat above applies)?
There was a problem hiding this comment.
Disregard, it can't be up to date because it is checked in the caller.
| public static final String NAME = FlushAction.NAME + "[s]"; | ||
| public static final ActionType<ReplicationResponse> TYPE = new ActionType<>(NAME); | ||
|
|
||
| protected final ProjectResolver projectResolver; |
| final List<BulkItemRequest> requests = entry.getValue(); | ||
|
|
||
| // Get effective shardCount for shardId and pass it on as parameter to new BulkShardRequest | ||
| var indexMetadata = project.getIndexSafe(shardId.getIndex()); |
There was a problem hiding this comment.
This is to replace the commented code below. The indexMetadata cannot really be null at this point and we must have a valid SplitShardCountSummary to pass to the BulkShardRequest.
There was a problem hiding this comment.
can we remove the commented code before pushing?
There was a problem hiding this comment.
Does this change the behavior in any way? I can't really image it does but i want to check.
There was a problem hiding this comment.
This is just to make sure we throw an assert if we do see null indexmetadata, which was not the case earlier. We never expected indexMetadata to be NULL in the first place so the behavior has not changed.
I will remove commented code.
| protected ShardFlushRequest newShardRequest(FlushRequest request, ShardId shardId, ProjectMetadata project) { | ||
| // Get effective shardCount for shardId and pass it on as parameter to new ShardFlushRequest | ||
| var indexMetadata = project.getIndexSafe(shardId.getIndex()); | ||
| SplitShardCountSummary reshardSplitShardCountSummary = SplitShardCountSummary.forIndexing(indexMetadata, shardId.getId()); | ||
| return new ShardFlushRequest(request, shardId, reshardSplitShardCountSummary); |
There was a problem hiding this comment.
I might have been tempted to modify TransportBroadcastReplicationAction.shards to return a list of shardid, summary pairs, to keep the binding between the shards we've chosen for routing and the summary that is calculated based on that routing table explicit. It doesn't address any kind of bug here, it just makes the API a little clearer and harder to use incorrectly.
There was a problem hiding this comment.
Yes that makes sense. Made the changes in latest upload.
| // We are here because there was mismatch between the SplitShardCountSummary in the request | ||
| // and that on the primary shard node. We assume that the request is exactly 1 reshard split behind | ||
| // the current state. | ||
| @Override | ||
| protected Map<ShardId, BasicReplicationRequest> splitRequestOnPrimary(BasicReplicationRequest request) { | ||
| ProjectMetadata project = projectResolver.getProjectMetadata(clusterService.state()); | ||
| final ShardId sourceShard = request.shardId(); | ||
| IndexMetadata indexMetadata = project.getIndexSafe(request.shardId().getIndex()); | ||
| SplitShardCountSummary shardCountSummary = SplitShardCountSummary.forIndexing(indexMetadata, sourceShard.getId()); | ||
| Map<ShardId, BasicReplicationRequest> requestsByShard = new HashMap<>(); | ||
| requestsByShard.put(sourceShard, request); | ||
| // Create a request for original source shard and for each target shard. | ||
| // New requests that are to be handled by target shards should contain the | ||
| // latest ShardCountSummary. | ||
| int targetShardId = indexMetadata.getReshardingMetadata().getSplit().targetShard(sourceShard.id()); | ||
| ShardId targetShard = new ShardId(request.shardId().getIndex(), targetShardId); | ||
| requestsByShard.put(targetShard, new BasicReplicationRequest(targetShard, shardCountSummary)); | ||
| return requestsByShard; | ||
| } | ||
|
|
||
| @Override | ||
| protected Tuple<ReplicationResponse, Exception> combineSplitResponses( | ||
| BasicReplicationRequest originalRequest, | ||
| Map<ShardId, BasicReplicationRequest> splitRequests, | ||
| Map<ShardId, Tuple<ReplicationResponse, Exception>> responses | ||
| ) { | ||
| int failed = 0; | ||
| int successful = 0; | ||
| int total = 0; | ||
| List<ReplicationResponse.ShardInfo.Failure> failures = new ArrayList<>(); | ||
|
|
||
| // If the action fails on either one of the shards, we return an exception. | ||
| // Case 1: Both source and target shards return a response: Add up total, successful, failures | ||
| // Case 2: Both source and target shards return an exception : return exception | ||
| // Case 3: One shards returns a response, the other returns an exception : return exception | ||
| for (Map.Entry<ShardId, Tuple<ReplicationResponse, Exception>> entry : responses.entrySet()) { | ||
| ShardId shardId = entry.getKey(); | ||
| Tuple<ReplicationResponse, Exception> value = entry.getValue(); | ||
| Exception exception = value.v2(); | ||
| if (exception != null) { | ||
| return new Tuple<>(null, exception); | ||
| } else { | ||
| ReplicationResponse response = value.v1(); | ||
| failed += response.getShardInfo().getFailed(); | ||
| successful += response.getShardInfo().getSuccessful(); | ||
| total += response.getShardInfo().getTotal(); | ||
| Collections.addAll(failures, response.getShardInfo().getFailures()); | ||
| } | ||
| } | ||
| ReplicationResponse.ShardInfo.Failure[] failureArray = failures.toArray(new ReplicationResponse.ShardInfo.Failure[0]); | ||
| assert failureArray.length == failed; | ||
| ReplicationResponse.ShardInfo shardInfo = ReplicationResponse.ShardInfo.of(total, successful, failureArray); | ||
| ReplicationResponse response = new ReplicationResponse(); | ||
| response.setShardInfo(shardInfo); | ||
| return new Tuple<>(response, null); | ||
| } |
There was a problem hiding this comment.
Can we share this logic with the flush action instead of duplicating it? That way e.g. fixing the possible stale cluster read happens in one place.
There was a problem hiding this comment.
Done. Created ReplicationRequestSplitHelper for this.
bcully
left a comment
bcully
left a comment
There was a problem hiding this comment.
Looks pretty good. How are we tracking the TODOs though?
| final List<BulkItemRequest> requests = entry.getValue(); | ||
|
|
||
| // Get effective shardCount for shardId and pass it on as parameter to new BulkShardRequest | ||
| var indexMetadata = project.getIndexSafe(shardId.getIndex()); |
There was a problem hiding this comment.
can we remove the commented code before pushing?
|
|
||
| try (var refs = new RefCountingRunnable(() -> finish(listener))) { | ||
| for (final ShardId shardId : shards) { | ||
| // for (final ShardId shardId : shards) { |
| /* | ||
| protected List<ShardId> shards(Request request, ProjectState projectState) { | ||
| assert Transports.assertNotTransportThread("may hit all the shards"); | ||
| List<ShardId> shardIds = new ArrayList<>(); | ||
| OperationRouting operationRouting = clusterService.operationRouting(); | ||
| String[] concreteIndices = indexNameExpressionResolver.concreteIndexNames(projectState.metadata(), request); | ||
| ProjectMetadata project = projectState.metadata(); | ||
| String[] concreteIndices = indexNameExpressionResolver.concreteIndexNames(project, request); | ||
| for (String index : concreteIndices) { | ||
| Iterator<IndexShardRoutingTable> iterator = operationRouting.allWritableShards(projectState, index); | ||
| var indexMetadata = project.index(index); | ||
| while (iterator.hasNext()) { | ||
| shardIds.add(iterator.next().shardId()); | ||
| ShardId shardId = iterator.next().shardId(); | ||
| SplitShardCountSummary reshardSplitShardCountSummary = SplitShardCountSummary.forIndexing(indexMetadata, shardId.getId()); | ||
| shardIds.add(shardId); | ||
| } | ||
| } | ||
| return shardIds; | ||
| } |
There was a problem hiding this comment.
probably better not to commit all this commented out code
There was a problem hiding this comment.
Yes sorry, I will remove it.
| /** | ||
| * @return all shard ids the request should run on | ||
| */ | ||
| protected List<Tuple<ShardId, SplitShardCountSummary>> shards(Request request, ProjectState projectState) { |
There was a problem hiding this comment.
this is fine but I do personally generally prefer records to tuples to make it easier to read the code where the result is consumed.
There was a problem hiding this comment.
Sure I can switch to using a record
| public ShardFlushRequest(FlushRequest request, ShardId shardId, SplitShardCountSummary reshardSplitShardCountSummary) { | ||
| super(shardId, reshardSplitShardCountSummary); | ||
| this.request = request; | ||
| this.waitForActiveShards = ActiveShardCount.NONE; // don't wait for any active shards before proceeding, by default |
There was a problem hiding this comment.
nit: extract this into a constant and share it between two constructors
| public static final String SOURCE_API = "api"; | ||
|
|
||
| private final Executor refreshExecutor; | ||
| protected final ProjectResolver projectResolver; |
There was a problem hiding this comment.
| protected final ProjectResolver projectResolver; | |
| private final ProjectResolver projectResolver; |
| })); | ||
| } | ||
|
|
||
| // We are here because there was mismatch between the SplitShardCountSummary in the request |
There was a problem hiding this comment.
Should this comment be on the method in the super class?
There was a problem hiding this comment.
Sure, that should make it more clear.
| final List<BulkItemRequest> requests = entry.getValue(); | ||
|
|
||
| // Get effective shardCount for shardId and pass it on as parameter to new BulkShardRequest | ||
| var indexMetadata = project.getIndexSafe(shardId.getIndex()); |
There was a problem hiding this comment.
Does this change the behavior in any way? I can't really image it does but i want to check.
| /** | ||
| * Creates a new request with resolved shard id | ||
| */ | ||
| // TODO: Check if callers of this need to be modified to pass in shardCountSummary |
There was a problem hiding this comment.
I think i would prefer for this constructor to not exist eventually because it's a "trap". If a particular call site does not need to pass summary, it can (with explanation).
| ShardId shardId = indexShard.shardId(); | ||
| if (request.refresh()) { | ||
| logger.trace("send refresh action for shard {}", shardId); | ||
| // TODO: Do we need to pass in shardCountSummary here ? |
There was a problem hiding this comment.
I guess we'd look at this while working on get? I think it's a stray todo in this PR though.
There was a problem hiding this comment.
This is covered by ES-13508. Sorry should have mentioned it here. I think I have tickets for all the TODOs.
4 participants
Address ES-13295 and ES-13170
Implement shard split logic for Flush and Refresh. Also forward the most recent shardCountRequestSummary to the target shards, when splitting the request at the source.