Introduce retention lease background sync #38262
There are no files selected for viewing
This file contains hidden or bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
This file contains hidden or bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
This file contains hidden or bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
This file contains hidden or bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
| Original file line number | Diff line number | Diff line change |
|---|---|---|
| @@ -0,0 +1,185 @@ | ||
| /* | ||
| * Licensed to Elasticsearch under one or more contributor | ||
| * license agreements. See the NOTICE file distributed with | ||
| * this work for additional information regarding copyright | ||
| * ownership. Elasticsearch licenses this file to you under | ||
| * the Apache License, Version 2.0 (the "License"); you may | ||
| * not use this file except in compliance with the License. | ||
| * You may obtain a copy of the License at | ||
| * | ||
| * http://www.apache.org/licenses/LICENSE-2.0 | ||
| * | ||
| * Unless required by applicable law or agreed to in writing, | ||
| * software distributed under the License is distributed on an | ||
| * "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY | ||
| * KIND, either express or implied. See the License for the | ||
| * specific language governing permissions and limitations | ||
| * under the License. | ||
| */ | ||
|
|
||
| package org.elasticsearch.index.seqno; | ||
|
|
||
| import org.apache.logging.log4j.LogManager; | ||
| import org.apache.logging.log4j.Logger; | ||
| import org.apache.logging.log4j.message.ParameterizedMessage; | ||
| import org.apache.lucene.store.AlreadyClosedException; | ||
| import org.elasticsearch.ExceptionsHelper; | ||
| import org.elasticsearch.action.ActionListener; | ||
| import org.elasticsearch.action.support.ActionFilters; | ||
| import org.elasticsearch.action.support.replication.ReplicationRequest; | ||
| import org.elasticsearch.action.support.replication.ReplicationResponse; | ||
| import org.elasticsearch.action.support.replication.TransportReplicationAction; | ||
| import org.elasticsearch.cluster.action.shard.ShardStateAction; | ||
| import org.elasticsearch.cluster.metadata.IndexNameExpressionResolver; | ||
| import org.elasticsearch.cluster.service.ClusterService; | ||
| import org.elasticsearch.common.inject.Inject; | ||
| import org.elasticsearch.common.io.stream.StreamInput; | ||
| import org.elasticsearch.common.io.stream.StreamOutput; | ||
| import org.elasticsearch.common.settings.Settings; | ||
| import org.elasticsearch.common.util.concurrent.ThreadContext; | ||
| import org.elasticsearch.index.shard.IndexShard; | ||
| import org.elasticsearch.index.shard.IndexShardClosedException; | ||
| import org.elasticsearch.index.shard.ShardId; | ||
| import org.elasticsearch.indices.IndicesService; | ||
| import org.elasticsearch.threadpool.ThreadPool; | ||
| import org.elasticsearch.transport.TransportService; | ||
|
|
||
| import java.io.IOException; | ||
| import java.util.Objects; | ||
|
|
||
| /** | ||
| * Replication action responsible for background syncing retention leases to replicas. This action is deliberately a replication action so | ||
| * that if a replica misses a background retention lease sync then that shard will not be marked as stale. We have some tolerance for a | ||
| * shard copy missing renewals of retention leases since the background sync interval is much smaller than the expected lifetime of | ||
| * retention leases. | ||
| */ | ||
| public class RetentionLeaseBackgroundSyncAction extends TransportReplicationAction< | ||
| RetentionLeaseBackgroundSyncAction.Request, | ||
| RetentionLeaseBackgroundSyncAction.Request, | ||
| ReplicationResponse> { | ||
|
|
||
| public static String ACTION_NAME = "indices:admin/seq_no/retention_lease_background_sync"; | ||
|
|
||
| private static final Logger LOGGER = LogManager.getLogger(RetentionLeaseSyncAction.class); | ||
|
|
||
| protected Logger getLogger() { | ||
| return LOGGER; | ||
| } | ||
|
|
||
| @Inject | ||
| public RetentionLeaseBackgroundSyncAction( | ||
| final Settings settings, | ||
| final TransportService transportService, | ||
| final ClusterService clusterService, | ||
| final IndicesService indicesService, | ||
| final ThreadPool threadPool, | ||
| final ShardStateAction shardStateAction, | ||
| final ActionFilters actionFilters, | ||
| final IndexNameExpressionResolver indexNameExpressionResolver) { | ||
| super( | ||
| settings, | ||
| ACTION_NAME, | ||
| transportService, | ||
| clusterService, | ||
| indicesService, | ||
| threadPool, | ||
| shardStateAction, | ||
| actionFilters, | ||
| indexNameExpressionResolver, | ||
| Request::new, | ||
| Request::new, | ||
| ThreadPool.Names.MANAGEMENT); | ||
| } | ||
|
|
||
| /** | ||
| * Background sync the specified retention leases for the specified shard. | ||
| * | ||
| * @param shardId the shard to sync | ||
| * @param retentionLeases the retention leases to sync | ||
| */ | ||
| public void backgroundSync( | ||
| final ShardId shardId, | ||
| final RetentionLeases retentionLeases) { | ||
| Objects.requireNonNull(shardId); | ||
| Objects.requireNonNull(retentionLeases); | ||
| final ThreadContext threadContext = threadPool.getThreadContext(); | ||
| try (ThreadContext.StoredContext ignore = threadContext.stashContext()) { | ||
| // we have to execute under the system context so that if security is enabled the sync is authorized | ||
| threadContext.markAsSystemContext(); | ||
| execute( | ||
| new Request(shardId, retentionLeases), | ||
| ActionListener.wrap( | ||
| r -> {}, | ||
| e -> { | ||
| if (ExceptionsHelper.unwrap(e, AlreadyClosedException.class, IndexShardClosedException.class) == null) { | ||
| getLogger().warn(new ParameterizedMessage("{} retention lease background sync failed", shardId), e); | ||
| } | ||
| })); | ||
| } | ||
| } | ||
|
|
||
| @Override | ||
| protected PrimaryResult<Request, ReplicationResponse> shardOperationOnPrimary(final Request request, final IndexShard primary) { | ||
| Objects.requireNonNull(request); | ||
| Objects.requireNonNull(primary); | ||
| primary.afterWriteOperation(); | ||
|
Member
There was a problem hiding this comment. I am not sure if this call is useful for it does not change translog nor Lucene. |
||
| return new PrimaryResult<>(request, new ReplicationResponse()); | ||
| } | ||
|
|
||
| @Override | ||
| protected ReplicaResult shardOperationOnReplica(final Request request, final IndexShard replica){ | ||
| Objects.requireNonNull(request); | ||
| Objects.requireNonNull(replica); | ||
| replica.updateRetentionLeasesOnReplica(request.getRetentionLeases()); | ||
| replica.afterWriteOperation(); | ||
|
Member
There was a problem hiding this comment. same here - I am not sure if this call is useful for it does not change translog nor Lucene.
Member
Author
There was a problem hiding this comment. The goal is to trigger periodic flush. Let us discuss the exact mechanics of this in a follow-up as that work develops. |
||
| return new ReplicaResult(); | ||
| } | ||
|
|
||
| public static final class Request extends ReplicationRequest<Request> { | ||
|
|
||
| private RetentionLeases retentionLeases; | ||
|
|
||
| public RetentionLeases getRetentionLeases() { | ||
| return retentionLeases; | ||
| } | ||
|
|
||
| public Request() { | ||
|
|
||
| } | ||
|
|
||
| public Request(final ShardId shardId, final RetentionLeases retentionLeases) { | ||
| super(Objects.requireNonNull(shardId)); | ||
| this.retentionLeases = Objects.requireNonNull(retentionLeases); | ||
| } | ||
|
|
||
| @Override | ||
| public void readFrom(final StreamInput in) throws IOException { | ||
| super.readFrom(in); | ||
| retentionLeases = new RetentionLeases(in); | ||
| } | ||
|
|
||
| @Override | ||
| public void writeTo(final StreamOutput out) throws IOException { | ||
| super.writeTo(Objects.requireNonNull(out)); | ||
| retentionLeases.writeTo(out); | ||
| } | ||
|
|
||
| @Override | ||
| public String toString() { | ||
| return "Request{" + | ||
| "retentionLeases=" + retentionLeases + | ||
| ", shardId=" + shardId + | ||
| ", timeout=" + timeout + | ||
| ", index='" + index + '\'' + | ||
| ", waitForActiveShards=" + waitForActiveShards + | ||
| '}'; | ||
| } | ||
|
|
||
| } | ||
|
|
||
| @Override | ||
| protected ReplicationResponse newResponseInstance() { | ||
| return new ReplicationResponse(); | ||
| } | ||
|
|
||
| } | ||
This file contains hidden or bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
This file contains hidden or bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
Oops, something went wrong.
Add this suggestion to a batch that can be applied as a single commit.
This suggestion is invalid because no changes were made to the code.
Suggestions cannot be applied while the pull request is closed.
Suggestions cannot be applied while viewing a subset of changes.
Only one suggestion per line can be applied in a batch.
Add this suggestion to a batch that can be applied as a single commit.
Applying suggestions on deleted lines is not supported.
You must change the existing code in this line in order to create a valid suggestion.
Outdated suggestions cannot be applied.
This suggestion has been applied or marked resolved.
Suggestions cannot be applied from pending reviews.
Suggestions cannot be applied on multi-line comments.
Suggestions cannot be applied while the pull request is queued to merge.
Suggestion cannot be applied right now. Please check back later.
Uh oh!
There was an error while loading. Please reload this page.