-
Notifications
You must be signed in to change notification settings - Fork 24.7k
Commit
This commit does not belong to any branch on this repository, and may belong to a fork outside of the repository.
Retention lease background sync 6.x (#38263)
This commit introduces a background sync for retention leases. The idea here is that we do a heavyweight sync when adding a new retention lease, and then periodically we want to background sync any retention lease renewals to the replicas. As long as the background sync interval is significantly lower than the extended lifetime of a retention lease, it is okay if from time to time a replica misses a sync (it will still have an older version of the lease that is retaining more data as we assume that renewals do not decrease the retaining sequence number). There are two follow-ups that will come after this commit. The first is to address the fact that we have not adapted the should periodically flush logic to possibly flush the retention leases. We want to do something like flush if we have not flushed in the last five minutes and there are renewed retention leases since the last time that we flushed. An additional follow-up will remove the syncing of retention leases when a retention lease expires. Today this sync could be invoked in the background by a merge operation. Rather, we will move the syncing of retention lease expiration to be done under the background sync. The background sync will use the heavyweight sync (write action) if a lease has expired, and will use the lightweight background sync (replication action) otherwise.
- Loading branch information
1 parent
e811b49
commit d5dd8db
Showing
20 changed files
with
662 additions
and
29 deletions.
There are no files selected for viewing
This file contains 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 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 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
185 changes: 185 additions & 0 deletions
185
server/src/main/java/org/elasticsearch/index/seqno/RetentionLeaseBackgroundSyncAction.java
This file contains 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(); | ||
| 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(); | ||
| 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 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 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.