Add Sequence Numbers to write operations

[bleskes]

Introduction

An Elasticsearch shard can receive indexing, update, and delete commands. Those changes are applied first on the primary shard, maintaining per doc semantics and are then replicated to all the replicas. All these operations happen concurrently. While we maintain ordering on a per doc basis, using versioning support there is no way to order them with respect to each other. Having such a per shard operation ordering will enable us to implement higher level features such as Changes API (follow changes to documents in a shard and index) and Reindexing API (take all data from a shard and reindex it into another, potentially mutating the data). Internally we could use this ordering to speed up shard recoveries, by identifying which specific operations need to be replayed to the recovering replica instead of falling back to a file based sync.

To get such ordering, each operation will be assigned a unique and ever increasing Sequence Number (in short, seq#). This sequence number will be assigned on the primary and replicated to all replicas. Seq# are to be indexed in Lucene to allow sorting, range filtering etc.

Warning, research ahead

What follows in this ticket is the current thinking about how to best implement this feature. It may change in subtle or major ways as the work continues. Is is important to implement this infrastructure in a way that is correct, resilient to failures, and without slowing down indexing speed. We feel confident with the approach described below, but we may have to backtrack or change the approach completely.

What is a Sequence

Applying an operation order on a primary is a simple question of incrementing a local counter for every operation. However, this is not sufficient to guarantee global uniqueness and monotonicity under error conditions where the primary shard can be isolated by a network partition. For those, the identity of the current primary needs to be baked into each operation. For example, late to arrive operations from an old primary can be detected and rejected.

In short, each operation is assigned two numbers:

• a term - this number is incremented with every primary assignment and is determined by the cluster master. This is very similar to the notion of a term in Raft, a view-number in Viewstamped Replication or an epoch in Zab.
• a seq# - this number is incremented by the primary with each operation it processes.

To achieve ordering, when comparing two operations , o1 & o2, we say that o1 < o2 if and only if s1.seq# < s2.seq# or (s1.seq# == s2.seq# and s1.term < s2.term). Equality and greater than are defined in a similar fashion.

For reasons explained later on, we maintain for each shard copy two special seq#:

1. local checkpoint# - this is the highest seq# for which all lower seq# have been processed . Note that this is not the highest seq# the shard has processed due to the concurrent indexing, which means that some changes can be processed while previous more heavy ones can still be on going.
2. global checkpoint# (or just checkpoint#) - the highest seq# for which the local shard can guarantee that all previous (included) seq# have been processed on all active shard copies (i.e., primary and replicas).

Those two numbers will be maintained in memory but also persisted in the metadata of every lucene commit.

Changes to indexing flow on primaries

Here is a sketch of the indexing code on primaries. Much of it is identical to the current logic. Changes or additions are marked in bold .

1. Validate write consistency based on routing tables.
2. Incoming indexing request is parsed first (rejected upon mapping/parsing failures)
3. Under uid lock:
   1. Versioning is resolved to a fixed version to be indexed.
   2. Operation is assigned a seq# and a term
   3. Doc is indexed into Lucene.
   4. Doc is put into translog.
4. Replication
   1. Failures in step 3 above are also replicated (eg due to failure of lucene tokenization)
   2. Send docs to all assigned replicas.
   3. Replicas respond with their current local checkpoint#.
   4. When all respond (or have failed), send answer to client.
5. Checkpoint update:
   1. Update the global `checkpoint# to the highest seq# for which all active replicas have processed all lower seq# (inclusive). This is based on information received in 4.3 .
   2. If changed, send new global checkpoint# to replicas (can be folded into a heartbeat/next index req).

Changes to indexing flow on replicas

As above, this is sketch of the indexing code on replicas. Changes with the current logic are marked as bold.

1. Validate request
   1. Seq#'s term is >= locally known primary term.
2. Under uid lock:
   1. Index into Lucene if seq# > local copy and doesn't represent an error on primary.
   2. Add to local translog.
3. Respond with the current local checkpoint#

Global Checkpoint# increment on replicas

The primary advances its global checkpoint# based on its knowledge of its local and replica's local checkpoint#. Periodically it shares its knowledge with the replicas

1. Validate source:
   1. source's primary term is == locally known primary term.
2. Validate correctness:
   1. Check that all sequence# below the new global checkpoint# were processed and local checkpoint# is of the same primary term. If not, fail shard.
3. Set the shard’s copy of global checkpoint#, if it's lower than the incoming global checkpoint.

Note that the global checkpoint is a local knowledge of that is update under the mandate of the primary. It may be that the primary information is lagging compared to a replica. This can happen when a replica is promoted to a primary (but still has stale info).

First use case - faster replica recovery

Have an ordering of operations allows us to speed up the recovery process of an existing replica and synchronization with the primary. At the moment, we do file based sync which typically results in over-copying of data. Having a clearly marked checkpoint# allows us to limit the sync operation to just those documents that have changed subsequently. In many cases we expect to have no documents to sync at all. This improvement will be tracked in a separate issue.

Road map

Basic infra

• Introduce Primary Terms (Introduce Primary Terms #14062)
• Introduce Seq# and index them (Add Sequence Numbers and enforce Primary Terms #14651)
• Introduce local checkpoints (Introduce Local checkpoints #15390)
• Introduce global checkpoints (Introduced Global checkpoints for Sequence Numbers #15485)
• Replicated failed operations for which a seq# was assigned (@areek) Replicate write failures #23314
• Create testing infrastructure to allow testing replication as a unit test, but with real IndexShards (Introduce Replication unit tests using real shards #18930)
• Persist local and global checkpoints (Persist sequence number checkpoints #18949)
• Update to Lucene 6.0 (Sequence numbers commit data for Lucene uses Iterable interface #20793)
• Persist global checkpoint in translog commits (@jasontedor) Add global checkpoint to translog checkpoints #21254
• Reading global checkpoint from translog should be part of translog recovery (@jasontedor) Clarify global checkpoint recovery #21934
• transfer global checkpoint after peer recovery (@jasontedor) Tighten sequence numbers recovery #22212
• add translog no-op (@jasontedor) Introduce translog no-op #22291
• Handle retry on primary exceptions in shard bulk action; we currently potentially reuse the answers of the previous primary. This is needed for versioned operations and mapping conflicts (we process ops, the third op needs a mapping change, we reach out to the master, the master already processed that change from another shard and responds with nothing to do, we reparse the op but the node doesn't have the mapping changes yet, we throw a retry and re-run the previous ops as well, potentiall running into conflicts) (@bleskes)
• Don't fail shards as stale if they fail a global check point sync - this will happen all the time when a primary is started and wants to sync the global checkpoint. To achieve this we agreed to change default behavior of shard failures for replication operations (with the exception of write operations, see later). If an operation fails on a replica, the primary shouldn't fail the replica or mark it as stale. Instead it should report this to the user by failing the entire operation. Write Replication Operation (i.e., sub classes of TransportWriteAction) should keep the current behavior. (@dakrone) Change certain replica failures not to fail the replica shard #22874
• Transfer sequence number "primary context" (i.e., in sync shards, tracked local checkpoints etc) upon primary relocation AND don't allow the old primary to operate as primary anymore (see IndexShard#verifyPrimary) (@jasontedor, Introduce primary context #25122)

Replica recovery (no rollback)

A best effort doc based replica recovery, based on local last commit. By best effort we refer to having no guarantees on the primary
translog state and the likelihood of doc based recovery to succeed and not requiring a file sync

• Move local checkpoint to max seq# in commit when opening engine Tighten sequence numbers recovery #22212
  We currently have no guarantee that all ops above the local checkpoint baked into the commit will be replayed. That means that delete operations with a seq# > local checkpoint will not be replayed. To work around it (for now), we will move the local checkpoint artificially (at the potential expense of correctness) (@jasontedor)
• Review correctness of POC and extract requirements for the primary side (@jasontedor) replaced with TLA+ work
• Use seq# checkpoints for replica recovery (Introduce sequence-number-based recovery #22484 , @jasontedor)

Translog seq# based API

Currently translog keeps all operations that are not persisted into the last lucene commit. This doesn't imply that it can serve all operations from a given seq# and up. We want to move seq# based recovery where a lucene commit indicates what seq# a fully baked into it and the translog recovers from there.

• Add min/max seq# to translog generations. (Introduce sequence-number-aware translog #22822, @jasontedor)
• Add a smaller maximum generation size and automatically create new generations. Note that the total translog size can still grow to 512MB. However, dividing this in smaller pieces will allow a lucene commit will be able to trim the translog, even though we may need some operations from the non-current generation (see next bullet point). (Introduce translog generation rolling #23606, @jasontedor)
• Move translog trimming to use lastCommittedLocalCheckpoint and use that for recovering operations when opening a lucene index. BWC aspects (a replica on new node operating without seq#) TBD. (Preserve multiple translog generations #24015, @jasontedor)
• Adapt snapshots to dedup sequence numbers by primary term (by reading in reverse) (Dedup translog operations by reading in reverse #27268, @dnhatn)
• Allow to trim all ops above a certain seq# with a term lower than X (Allow to trim all ops above a certain seq# with a term lower than X #30176 @vladimirdolzhenko)

Primary recovery (no rollback)

• After recovering from the translog, the primary should close any gaps in its history by indexing no ops into the missing seq# (@s1monw, Fill missing sequence IDs up to max sequence ID when recovering from store #24238)
• After recovering, a primary should update its knowledge of its own local checkpoint (@ywelsch, Introduce primary/replica mode for GlobalCheckPointTracker #25468)

Primary promotion

• Implement a clean transition between replica operations on the shard (with an older term) and operations with the new term.
  • Replicas (triggered by indexing operations and live sync) (Block older operations on primary term transition #24779 @jasontedor)
  • Primary (via a cluster state update) (Introduce clean transition on primary promotion #24925 @jasontedor)
• Should trigger a translog role over (so operations from the new primary will always have a higher translog generation than any operation they may conflict with) (Guarantee that translog generations are seqNo conflict free #24825 @bleskes)
• Should close gaps in history. Note that this doesn't imply this information will be transferred to the replicas. That is the job of the primary/replica sync. (Fill gaps on primary promotion #24945 @jasontedor)

Live replica/primary sync (no rollback)

• Add a task that streams all operations from the primary's global checkpoint to all shards. (@ywelsch Remove TranslogRecoveryPerformer #24858)

  • fails the shard on other failures (double think about shard closing)
  • triggers update of primary term on replica if needed
  • transfers shard's local checkpoint back to the new primary
  • progress reporting via the task manager infrastructure

• When a replica shard increases its primary term under the mandate of a new primary, it should also update its global checkpoint; this gives us the guarantee that its global checkpoint is at least as high as the new primary and gives a starting point for the primary/replica resync (@ywelsch, Update global checkpoint when increasing primary term on replica #25422)

• Replicas should throw their local checkpoint to the global checkpoint when detected a new primary. (@jasontedor, Throw back replica local checkpoint on new primary #25452)

Primary recovery with rollback

Needed to deal with discrepancies between translog and commit point that can result of failure during primary/replica sync

• Make sure the translog keeps data based on the oldest commit (Introducing a translog deletion policy #24950)
• A custom deletion policy to keep old commits around (@dnhatn, Keep commits and translog up to the global checkpoint #27606)
• Roll back before starting to recover from translog (lucene's deletion policy can clean commits before opening)(@dnhatn Rollback a primary before recovering from translog #27804)
• Clean up commits once they are no longer needed. Lucene only calls the deletion policy upon commit. Since we keep older commit around based on the advancement of the global checkpoint, we can trim data earlier, i.e., once the global checkpoint has advanced enough(@dnhatn Clean up commits when global checkpoint advanced #28140).

Replica recovery with rollback

Needed to throw away potential wrong doc versions that ended up in lucene. Those "wrong doc versions" may still be in the translog of the replica but since we ignore the translog on replica recovery they will be removed.

• Only use "valid" commit points as the basis for file based recovery (i.e., commit points below the global check point)(@dnhatn Primary send safe commit in file-based recovery #28038)
• Only use "valid" commit points as the basis for ops based recovery (lucene's deletion policy can clean commits before opening)(@dnhatn Replica starts peer recovery with safe commit #28181)

Live replica/primary sync with rollback

• Allow a shard to rollback to a seq# from before last known checkpoint# based on NRT readers
• Index all operations missing from the rollback point up to the global checkpoint from local translog

Seq# as versioning

• Change InternalEngine to to resolve collision based on seq# on replicas and recovery (@bleskes, Use sequence numbers to identify out of order delivery in replicas & recovery #24060)
• Change write API to allow specifying desired current seq# for the operation to succeed
• Make doc level versioning an opt in feature (mostly for external versioning)

Shrunk indices

Shrunk indices have mixed histories.

• The primary term of the new index should be higher than the source. (@jasontedor, Initialize primary term for shrunk indices #25307)
• When recovering the primary shard we need to set it's maximum seqNo to the max of all the source shards. (@jasontedor, Initialize sequence numbers on a shrunken index #25321)
• (PS - while we're at it we can also transfer the max max_unsafe_auto_id_timestamp @jasontedor, Initialize max unsafe auto ID timestamp on shrink #25356)

Adopt Me

• Properly store seq# in lucene: we expect to use the seq# for sorting, during collision checking and for doing range searches. The primary term will only be used during collision checking when the seq# of the two document copies is identical. Mapping this need to lucene means that the seq# it self should be stored both as a numeric doc value and as numerical indexed field (BKD). The primary term should be stored as a doc value field and doesn't need an indexed variant. We also considered the alternative of encoding both term and seq# into a single numeric/binary field as it may save on a the disk lookup implied by two separate fields. Since we expect the primary term to be rarely retrieved, we opted for the simplicity of the two doc value fields solution. We also expect it to mean better compression. (@dakrone) Add internal _primary_term field to store shard's primary term #21480
• Add primary term to DocWriteResponse (@jasontedor) Add primary term to doc write response #24171
• how do we deal with the BWC aspects in the case that - a primary is running on a new node will one replica is on an old node and one replica is on a new one. In that case the primary will maintain seq# and checkpoints for itself and the replica on the new node. However if the primary fails it may be the case that the old replica is elected as primary. That means that the other replica will suddenly stop receiving sequence numbers. It is not clear if this really a problem and if so what the best approach to solve it. (@dakrone, Promote replica on the highest version node #25277)
• Introduce new shard states to indicated an ongoing primary sync on promotion. See Live primary-replica resync (no rollback) #24841 (review) . We now have an alternative plan for this - see Introduce promoting index shard state #28004 (comment)
• Delay primary relocation if primary has not yet synced with replicas . See Live primary-replica resync (no rollback) #24841 (review)

TBD

• A primary that's allocated when in sync replicas aren't can advance the global checkpoint to a region that's unsafe - the primary doesn't about if it's local ops, which are above the global checkpoint actually exists on the other replicas (@ywelsch Introduce primary/replica mode for GlobalCheckPointTracker #25468).
• File based recovery (both local and remote) can recreate deleted docs. If a delete is out of order with a previous index operation, we may replay the index operation on recovery, but not the delete. This has to do with the fact that we trim the translog/capture a starting point at an arbitrary generation, and replay all of it. Current solution - change the indexing plan on non-primary origin to never index to lucene below the local checkpoint (5.x is addressed in Engine - Do not store operations that are not index into lucene in the translog (5.x only) #25592)
• How to deal with failures to sync the global check point? this is interesting as we will rely on the checkpoint to be eventually updated on all shard copies.
• Throwing back local checkpoint to global checkpoint may leave us in a situation we don't have all ops in the translog to do primary/replica sync. This is because we don't have (yet) any guarantee that the translog has all ops above the global checkpoint. That shard will be problematic when promoted to primary (it doesn't have the data to do a sync), causing replica's local checkpoints to get stuck. This will change when we have a custom deletion policy but we may want to double check that and fail the primary if it can't sync it's replica.
• When indexing stops, sync global checkpoint faster (now we wait on idle shard) (@jasontedor, Introduce global checkpoint background sync #26591)
• Make local check point storage lazy intitialized to protect against memory usage during recovery (@jasontedor, Lazy initialize checkpoint tracker bit sets #27179)
• Snapshot and restore may create a primary that violates all our checkpointing logic. We should quantify the scenarios this can happen and discuss appropriate solutions. A similar issue occurs with force allocate empty & stale primaries.
• When indexing on a replica InternalEngine loads the primary term to resolve a potential conflict between two indexing operations to the same doc with the same seq#. Once we have proper rollbacks that should never be needed. Instead we should assert that the term is identical (and potentially that the doc it self is identical).
• Make replica shard allocation code aware of sequence-numbers, so that it selects a node with an existing shard copy not purely based on the number of matching Lucene segments, but also based on the number of operations that it's missing from the primary. Similarly, do the same to cancel existing recoveries if a node with an existing shard copy comes back that has better matching stats. (@dnhatn, Sequence number based replica allocation #46959)

Completed Miscellaneous

• Review feasibility of old indices (done and implemented in Add BWC layer to seq no infra and enable BWC tests #22185 ) (@bleskes)
• Remove Shadow replicas (@dakrone Remove shadow replicas #23906)
• If minimum of all local checkpoints is less than global checkpoint on the primary, do we fail the shard? No, this can happen when replicas pull back their local checkpoints to their version of the global checkpoint
• Failed shards who's local checkpoint is lagging with more than 10000 (?) ops behind the primary . This is a temporary measure to allow merging into master without closing translog gaps during primary promotion on a live shard. Those will require the replicas to pick them up, which will take a replica/primary live sync

[shikhar]

First use case - faster replica recovery

I'd argue the first use case is making replication semantics more sound :)

[rkonda]

It's not clear as to what would happen in the following split brain scenario (scenario-1):

1. split occurs, forming two networks
2. the network that didn't have a master, elects a master (call this network-2)
3. the master will elect a new primary (in network-2)
4. the primary in network-2 now has incremented term value (say 11). The primary in network-1 continues to have the same term value (10 in this example)
5. The connection between the networks is re-established.

In this case we need a strategy for reconciling the differences in the indexes, if there were change operations in both the networks. Does a strategy like that exist today? So far it seems like this situation is preventable by using min_master_nodes. However in case min_master_nodes is not set appropriately, some default strategy should come into effect I would think.

An example strategy could be:

1. Keep logs of write operations in both networks for a configurable amount of time. If the networks' connectivity is restored within this time period: (a) Drop all nodes in network-2 to read-only replica status (b) Attempt to reconcile the differences, and use network-1's state if the differences are not reconcilable. (c) Remove read-only status
   If the connectivity isn't restored within that time, when connection is restored, all indices in network-2 that have competing primaries in network-1 will lose their shards, and replicas are created from network-1.

Another interesting situation (scenario-2) to consider:

1. Continuing with the scenario described above until (4) ...
2. network-1 has another split, creating network-1 and network-1a. Network-1a gives term value of 11 to the new primary in that network.
3. network-1 completely fails, and connectivity between network-1a and network-2 are restored. Now we may have a scenario where the subsequent change operations might not fail but still lead to different indexes in the replicas, with some operations failing some of the time, creating a messy situation.

This would happen if there is no reconciliation strategy in effect.

I do see that the sequence numbering method will keep shards that have connectivity to both the networks, in integral state, in the case of scenario-1. In the case of scenario-2, it is possible that the same shard gets operations with same term values from multiple primaries, and that again could create faulty index in that replica.

I am still trying to understand Elasticsearch's cluster behavior. It's possible that I might have made assumptions that aren't correct.

[bleskes]

In this case we need a strategy for reconciling the differences in the indexes, if there were change operations in both the networks. Does a strategy like that exist today?

The current strategy, which seq# will keep enforcing but in easier/faster way, is that all replicas are "reset" to be an exact copy of the primary currently chosen by the master. As you noted, this falls apart when there are two residing masters in the cluster. Indeed, the only way to prevent this is by setting minimum master nodes - which is the number one most important setting to set in ES (tell it what the expected cluster size is)

If min master nodes is not set and a split brain occurs, resolution will come when one of the masters steps down (either by manual intervention or by detecting the other one). In that case all replicas will "reset" to the primary designated by the left over master.

Drop all nodes in network-2 to read-only replica status

This is similar to what ES does - nodes with no master will only serve read requests and block writes (by default, it can be configured to block reads).

it is possible that the same shard gets operations with same term values from multiple primaries, and that again could create faulty index in that replica.

If the term is the same from both primaries, the replica will accept them according to the current plan. The situation will be resolved when the network restores and the left over primary and replica sync but indeed there are potential troubles there. I have some ideas on how to fix this specific secondary failure (split brain is the true issue, after which all bets are off) but there are bigger fish to catch first :)

[rkonda]

Thank you very much for your clarification. I rather enjoy all these discussions and your comments.

The current strategy, which seq# will keep enforcing but in easier/faster way, is that all replicas are "reset" to be an exact copy of the primary currently chosen by the master.

The situation will be resolved when the network restores and the left over primary and replica sync

I would like to clearly understand the reset/sync scenarios. What triggers reset/sync?

I can think of a couple of "normal" operation scenarios

1. I would think that whenever a node joins a network, the master would initiate a sync/reset.
2. If a replica fails for a request, I suppose the primary should keep attempting a sync/reset, otherwise the replica might keep diverging, and at some point the master has to decommission that replica, otherwise the reads would be inconsistent.

In the case of split brain, with multi-network replicas (assuming min master nodes is set), primary-1 has been assuming that this replica R (on this third node, say N-3) has been failing (because of its allegiance to primary-2 ) but still is in the network. Hence it would attempt sync/reset. How does this protocol work? Should master-1 attempt to decommission R at some point, going by assumption (2)?

This problem will occur in a loop if R is decommissioned but another replica is installed on N-3 in its place, by the same protocol. There will be contention on N-3 for "reset"-ing replica shards by both the masters.

I suppose one way to resolve this is by letting a node choose a master if there are multiple masters. If we did this, then whenever a node loses its master, it would choose the other master, and there will be a sync/reset and all is well.

However if the node chooses its master, the other master will lose quorum, and hence cease to exist, which is a good resolution for this issue in my opinion.

[bleskes]

The two issues you mention indeed trigger a primary/replica sync. I'm not sure I follow the rest, I would like to ask you to continue the discussion on discuss.elastic.co . We try to keep github for issues and work items. Thx!

[rkonda]

Sure. Posted it here: https://discuss.elastic.co/t/sequence-numbers-to-write-ops-split-brain-scenario/43748

[makeyang]

any plan to release this?
it seems after this release, u guys will make ES a AP system? will u provide config paramters to allow users to control ES to be a AP or CP system eventually?

[bleskes]

@makeyang this will be released as soon as it is done. There's still a lot of work to do.

it seems after this release, u guys will make ES a AP system? will u provide config paramters to allow users to control ES to be a AP or CP system eventually?

ES is currently and will stay CP in the foreseeable future. If a node is partitioned away from the cluster it will serve read requests (configurable) but will block writes, in which case we drop availability. Of course in future there are many options but currently there are no concrete plans to make it any different.

[bleskes]

The work prescribed in this issue now completed and will be part of the coming 6.7 and 7.0 releases. There are still some small follow ups we want to do, but they do no need to be tracked as part of this issue. We now consider this completed.