Allow to seal an index

[s1monw]

In a lot of use cases indices are used for indexing only for a limited amount of time. Ie. in the daily index use case indices are created with a highish number of shards to scale out indexing and then after a couple of days these indices are idle in terms of writing. Yet, we still keep all the resources open since we are accepting writes at any time. This not necessary in a lot of cases and would allow for a large amount of optimizations:

• we can close IndexWriter and all it's resources.
• primary promotions are very simple since no changes are coming in
• indices might be able to be merged ie. if you have 7 indices for per week we can allow people to
  merge them into a weekly index to reduce the number of indices.
• query caches can rely on the fact that the index doesn't change and we can pull out min/max values for numeric fields with more confidence that it's worth the effort.
• maybe most important, recoveries are instant. We would issue a special commit when we seal that ensures that all changes have been written containing some kind of identifier that we can use in recovery which will prevent the entire copying of files in situations of full cluster restarts etc. It will reduce the restart times dramatically with very low complexity for all the time-based indices cases which usually suffer from slow cluster restarts due to the large number of indices and shards

[kimchy]

++, I think the improved recovery part idea is a killer. I wonder if we can re-use one way or another the read only flag we have today on an index.

[bleskes]

nice

[s1monw]

@kimchy I think we can. To me it's a 2 staged process, first we switch the index read only using the flag we have and once we are there and the cluster state has been published we seal the index which causes the actual optimizations to happen. so I think we can just reuse it?

[kimchy]

@s1monw ++, exactly what I was thinking about, and we have the flag recorded in the cluster state, so might need to record it somewhere else as well to make sure nothing went in, ++.

[s1monw]

adding this is also going to make full-cluster restart as well as rolling restarts likely instant. Even for the non-timeseries data / logging case for full restarts we can seal all the indices, shutdown, restart & unseal. This also works for rolling restarts if folks can afford having read only indices which I think is reasonable im most cases since the restart will be pretty fast. very promising!

[runningman84]

+1

[nik9000]

This also works for rolling restarts if folks can afford having read only indices which I think is reasonable im most cases since the restart will be pretty fast. very promising!

This is great! Our use case wouldn't allow us to seal an index outside of a rolling restart window or some other temporary maintenance action but we can absolutely get away with sealing them all for an hour or so.

So this is a great solution for us!

[synhershko]

+1, especially if sealing is reversible (which #10032 (comment) implies)

[rtoma]

+100 for 'instant recoveries'

[bleskes]

@synhershko yeah. the plan is to allow to unseal as well, making it a viable upgrade/full restart strategy (if you can afford stopping indexing).

[s1monw]

This is great! Our use case wouldn't allow us to seal an index outside of a rolling restart window or some other temporary maintenance action but we can absolutely get away with sealing them all for an hour or so.

yes it's absolutely possible to unseal and the operation should be very fast. ie. makeing a cluster state update essentially.

[s1monw]

We had some internal discussions how to implement this and I wanted to make sure they are recorded here on the issue. Sealing an index happens basically on two levels, the index and the shard level.

Index Level sealing

On the index level we use a ClusterBlock to prevent any write operations on the index via index.blocks.read_only. This is basically a cluster state update that sets the write block on the index that will be sealed together with a seal_id the seal id is a token that is generated to identify a seal operation on a shard level and also at a later point in time to utilize during recovery. So essentially there are three values on a ClusterState for each sealed index:

• a index.blocks.read_only to prevent writes
• a seal state index.seal.state which can either be sealing or sealed
• a index.seal.id in IndexMetaData.settings

This also requires the entire cluster to be on a version that supports and understands index sealing otherwise this feature will not be available. (we have the ability to check this via DiscoveryNodes#smallestNonClientNodeVersion()). The seal id should not be updatable outside of the seal API while the index.blocks.read_only is. Yet we need to prevent that this can be changed while the index is sealed.

The seal process is essentially a cluster state update (setting the block and the id) that waits for all shards to respond. This is very similar to how deleting an index works today. We issue the cluster state update that subsequently gets propagated to all the nodes in the cluster. Inside IndicesClusterStateService we listen to the relevant changes and update the IndexService where applicable.

Once the seal operation is issued we set index.seal.state : sealing. The master now registers a listener on a dedicated endpoint waiting for all relevant shards to reply with successful or failed seal operations. Once all shards replied the master issues the index.seal.state : sealed and responds to the user unless we already ran into a timeout. (note this is a non-blocking operation on the master just like all other actions). In the case of a timeout or if the cluster is stuck in sealing mode only an unseal operation can recover from that state. Unsealing is a pretty straight forward it basically removes the seal state from the index settings and publishes the clusterstate.

This is also very similar to the delete logic which is currently implemented in IndicesClusterStateService#applyDeletedShards and NodeIndexDeletedAction

Shard Level sealing

Once the IndexService knows about the sealing it essentially needs to wait until all in-flight operations are finished on the shards primary as well on all the replicas.

Shard level sealing

For this we are currently planning to use ref-counting similar to what we do on Store.java. The ref-counting implemented in AbstractRefCoutned works in a way that prevents the caller from incrementing the reference count once it reached 0. We can utilize this and increment the counter once in the IndexShard itself and decrement it again once the shard is either closed or sealed. Once we decremented the shards reference we only need to wait until
we reach 0 on the counter in order to process the sealing.

The good news is that due to the cluster block (read only settings) no new indexing operations can be issues such that we will reach 0 eventually. Certainly this requires reference counting (incRef / decRef) in the relevant API users which might be likely reducible to TransportShardReplicationOperationAction.
Once we reached 0 on the ref-count we can executed the following actions:

• issue a seal-commit on the engine that writes the shards seal ID to the commit metadata just like we do with the transaction log ID with every commit. This action is complex and should wait for all recoveries to finish (or even cancel?), flush the transaction log etc. to ensure all changes are committed to the lucene index.
• switch the engine from InternalEngine to an engine that is read-only to reduce resource utilization. We might be able to reuse, modify or abstract ShadowEngine for this purpose.
• send a seal command to the replicas of the shard which basically repeats the three steps above.
• send a seal acknowledgement to the master to eventually return to the user

At that point the index is sealed and no write operation can be submitted to the index anymore.

Unseal operation

The unseal operation pretty much reverses the sealing. We process a cluster state update that marks the index as unsealed by removing the seal states and processes the update. Once the cluster state update was successfully published we remove the index block and the index is good to go.

On a shard level we basically re-increment the reference count to 1 (allowing incoming requests to increment the count as well) and before doing that switching to InternalEngine to allow writes. The first write to the engine basically invalidates the seal such that we can't use it anymore for other operations like recovery.

Fast Recovery

Today recovery is very resource heavy and often super slow since we don't know if two shards are identical on a document level ie. did all operations reach the replica or not. We can tell on a lucene segment level but the segments are different on all replicas unless we copied the over which takes a huge amount of time. With index sealing we basically mark the replicas as identical on a operation / document level since we prevent all writes on the shards. That allows us to side-step the entire shard copying and startup replicas immediately.

Luckily implementing fast recovery on top of the sealing is very straight forward. Basically what we need is an extension of the RecoverySourceHandler or Engine.RecoveryHandler that can sidestep the entire recovery process if the seal-ID matches on both the replica and the primary and if the recovery source has no operations in it's translog. If this applies we can simply skip the entire heavy part of the recovery procedure and startup the replica immediately. We only need to be carful to not remove the seal ID from the indices by issuing commits at the end of the recovery.

For safety reasons, if any operations exist in the transaction log we can't utilize the seal ID for fast recovery. Any operation in the translog indicates an illegal state in the context of the seal ID or in other words it breaks the seal. For instance if an old replica is started on a node that was sealed before but the primary is already accepting writes again we can in theory only recover from the transaction log but for the initial iteration we should skip this optimization. In the future we might be even able to extend this process to issue seal commits on a per shard level while accepting writes.

Optimizing / Force Merge on a Sealed index

For the time based indices usecase it's important to run force-merge / optimize on these indices even if they are sealed. As an later extension we can allow users to run these operations even on a read-only engine. This is a feature that can be implemented at a later stage.

Proposed work items

• add ref-counting to IndexShard and use it in index modifying operations on ie. on TransportShardReplicationOperationAction This should be nicely testable and can be implemented entirely stand-alone
• implement a Engine#seal(String sealId) that flushes the translog, writes the seal id to the lucene index. This operation should fail if there is a recovery in progress and should maybe even close the engine forcefully. The later integration of this action should be straight forward since we can open a read-only engine next to the InternalEngine, swap the read-write engine out, seal it and refresh the read-only engine.
• implement a seal action on IndexShard that utilize the Engine#seal as stated above.
• implement a TransportShardReplicationOperationAction like class that runs seal commands on all replicas of a shard
• implement sealing on the master
• implement un-sealing on the master
• use seal ID in recovery. In the first iteration we only use it if there is no operation in the translog. Future iterations might allow to recover from translog as well if the seal ID is the same in the commit point on primary and replica.

I hope I covered all the moving parts at least on a high level. if there are any questions feel free to ask. Once we basically agree I will move this to the issue itself.

[andrassy]

+1

[bleskes]

Discussing this we came up with a new and simpler plan, which works independently of the cluster update. This gist of it is to have a best effort operation to sync the commit points both primaries and replicas. This "synced flush" is guaranteed to succeed if there are no concurrent indexing operations but will fail gracefully if there are. The result is a marker (sync id) on lucene commit points which allows us to shortcut the phase1 of recoveries which will give us the desired speed up. Since this is a best effort approach we can trigger it when ever a shard becomes inactive or in regular, longish intervals (say 30m) or any other time (TBD).

Solution sketch (this is a shard operation):

1. Pre Synced Commit phase:
   1. reaches out to all assigned shard copies and flush them.
   2. returns the lucene commit ID resulting of this flush.
2. Validate there are no inflight operations (if there are, we abort). This can be done using the request ref count described above.
3. Synced Commit phase:
   1. Generate a sync id (a GUID).
   2. On primary:
      1. If there are pending writes or the lucene commit id is not identical to the one retrieved by the pre sync phase, abort.
      2. Create a new commit point with the sync id (while blocking writes to make sure nothing slips in)
   3. On replicas, we repeat the steps done on the primary. Note that if a replica doesn't participate in the sync it's OK.

TODOs:

• Introduce ref counting of TransportShardReplicationOperation Ref count write operations on IndexShard #10610
• Implement preSyncCommit action based on TransportShardSingleOperationAction Actions for the two phases of synced flush  #10732 and afdab84
• Implement shard level IndexShard.syncIfNoPendingChanges(syncId, expectedCommitId) to be used by step 3 above. ebeb324
• implement Synced Commit action base on TransportShardReplicationOperation to do step 3. The request of this action will carry the relevant shard lucene commit ids. Actions for the two phases of synced flush  #10732 and afdab84
• extend the shard stats API to return the current Commit Data from lucene. This will allow inspecting the current sync id on all shards. Add CommitStats to supply information about the current commit point #10687
• change recovery code to shortcut phase1 if the source and target sync id are the same. use sync id when recovering #10775
• implement check for pending writes based on index shard ref counter (3.2)
• hook synced flush into IndexingMemoryController
• api for manually triggering a sync flush

[x] -> in feature branch https://github.com/elastic/elasticsearch/tree/feature/synced_flush

[s1monw]

thx for updating @bleskes

[s1monw]

@brwe can we close this one?

[bleskes]

a short note that this has been in implemented as synced flush - see #11179 & #11336 for more info