{"action":"created","issue":{"url":"https://api.github.com/repos/elastic/elasticsearch/issues/10708","repository_url":"https://api.github.com/repos/elastic/elasticsearch","labels_url":"https://api.github.com/repos/elastic/elasticsearch/issues/10708/labels{/name}","comments_url":"https://api.github.com/repos/elastic/elasticsearch/issues/10708/comments","events_url":"https://api.github.com/repos/elastic/elasticsearch/issues/10708/events","html_url":"https://github.com/elastic/elasticsearch/issues/10708","id":69958694,"number":10708,"title":"Add Sequence Numbers to write operations","user":{"login":"bleskes","id":1006375,"avatar_url":"https://avatars.githubusercontent.com/u/1006375?v=3","gravatar_id":"","url":"https://api.github.com/users/bleskes","html_url":"https://github.com/bleskes","followers_url":"https://api.github.com/users/bleskes/followers","following_url":"https://api.github.com/users/bleskes/following{/other_user}","gists_url":"https://api.github.com/users/bleskes/gists{/gist_id}","starred_url":"https://api.github.com/users/bleskes/starred{/owner}{/repo}","subscriptions_url":"https://api.github.com/users/bleskes/subscriptions","organizations_url":"https://api.github.com/users/bleskes/orgs","repos_url":"https://api.github.com/users/bleskes/repos","events_url":"https://api.github.com/users/bleskes/events{/privacy}","received_events_url":"https://api.github.com/users/bleskes/received_events","type":"User","site_admin":false},"labels":[{"url":"https://api.github.com/repos/elastic/elasticsearch/labels/:Sequence%20IDs","name":":Sequence IDs","color":"009800"},{"url":"https://api.github.com/repos/elastic/elasticsearch/labels/feature","name":"feature","color":"328a58"},{"url":"https://api.github.com/repos/elastic/elasticsearch/labels/Meta","name":"Meta","color":"e11d21"},{"url":"https://api.github.com/repos/elastic/elasticsearch/labels/resiliency","name":"resiliency","color":"009800"}],"state":"open","locked":false,"assignee":null,"milestone":null,"comments":2,"created_at":"2015-04-21T21:37:04Z","updated_at":"2016-03-08T06:53:13Z","closed_at":null,"body":"Introduction\r\n----\r\n\r\nAn 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](http://www.elastic.co/guide/en/elasticsearch/reference/current/docs-index_.html#index-versioning) 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.\r\n\r\nTo 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.\r\n\r\nWarning, research ahead\r\n-----------------------\r\n\r\nWhat 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. \r\n\r\n\r\nWhat is a Sequence #\r\n---------\r\n\r\nApplying 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.\r\n\r\nIn short, each seq# consists of two numbers:\r\n\r\n* 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.\r\n* a  `counter` - this number is incremented by the primary with each operation it processes.\r\n\r\n\r\nTo achieve ordering, when comparing two seq# , `s1` & `s2`, we say that `s1` < `s2` if and only if `s1.term` < `s2.term` or (`s1.term` == `s2.term` and `s1.counter` < `s2.counter`). Equality and greater than are defined in a similar fashion.\r\n\r\nFor reasons explained later on, we maintain for each shard copy two special seq#:\r\n\r\n1. `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.\r\n\r\n2. `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).\r\n\r\nThose two numbers will be maintained in memory but also persisted in the metadata of every lucene commit. \r\n\r\nChanges to indexing flow on primaries\r\n------------\r\n\r\nHere 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** .\r\n\r\n1. Validate write consistency based on routing tables.\r\n2. Incoming indexing request is parsed first (rejected upon mapping/parsing failures)\r\n3. Under uid lock:\r\n\t1. Versioning is resolved to a fixed version to be indexed.\r\n\t2. **Operation is assigned a seq#**\r\n\t3. Doc is indexed into Lucene. \r\n\t4. Doc is put into translog. \r\n4. Replication\r\n\t1. **Failures in step 3 above are also replicated (eg due to failure of lucene tokenization)**\r\n\t2. Send docs to all assigned replicas.\r\n\t3. **Replicas respond with their current `local checkpoint#`.**\r\n\t4. When all respond (or have failed), send answer to client.\r\n5. **Checkpoint update**:\r\n\t1. 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 . \r\n\t2. If changed, send new global `checkpoint#` to replicas (can be folded into a heartbeat/next index req).\r\n\r\nChanges to indexing flow on replicas\r\n------------------\r\n\r\nAs above, this is sketch of the indexing code on replicas. Changes with the current logic are marked as bold.\r\n\r\n1. **Validate request**\r\n\t1. Seq#'s term is >= locally known primary term.\r\n2. Under uid lock:\r\n\t2. Index into Lucene **if seq# > local copy and doesn't represent an error on primary**.\r\n\t3. Add to local translog.\r\n3. Respond with the current `local checkpoint#`\r\n\r\n\r\n\r\nGlobal Checkpoint# increment on replicas\r\n----\r\n\r\nThe 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\r\n\r\n1. Validate source:\r\n\t1. Seq#'s primary term is == locally known primary term.\r\n2. Validate correctness:\r\n\t3. Check that  all sequence# below the new `global checkpoint#` were processed and local checkpoint# is of the same primary term. If not, fail shard.\r\n2. Set the shard’s copy of `global checkpoint#`\r\n\r\n\r\nFirst use case - faster replica recovery\r\n--------------\r\n\r\nHave 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. \r\n\r\n\r\nRoad map\r\n----\r\n- [x] Introduce Primary Terms (#14062)\r\n- [x] Introduce Seq# and index them (#14651)\r\n- [ ] Introduce checkpoint # (both local and global)\r\n- [ ] Allow a shard to rollback to a seq# from before last known checkpoint#\r\n- [ ] Use seq# in recovery & primary failures (resync replicas with new primaries)\r\n- [ ] Documentation"},"comment":{"url":"https://api.github.com/repos/elastic/elasticsearch/issues/comments/193632541","html_url":"https://github.com/elastic/elasticsearch/issues/10708#issuecomment-193632541","issue_url":"https://api.github.com/repos/elastic/elasticsearch/issues/10708","id":193632541,"user":{"login":"rkonda","id":1133338,"avatar_url":"https://avatars.githubusercontent.com/u/1133338?v=3","gravatar_id":"","url":"https://api.github.com/users/rkonda","html_url":"https://github.com/rkonda","followers_url":"https://api.github.com/users/rkonda/followers","following_url":"https://api.github.com/users/rkonda/following{/other_user}","gists_url":"https://api.github.com/users/rkonda/gists{/gist_id}","starred_url":"https://api.github.com/users/rkonda/starred{/owner}{/repo}","subscriptions_url":"https://api.github.com/users/rkonda/subscriptions","organizations_url":"https://api.github.com/users/rkonda/orgs","repos_url":"https://api.github.com/users/rkonda/repos","events_url":"https://api.github.com/users/rkonda/events{/privacy}","received_events_url":"https://api.github.com/users/rkonda/received_events","type":"User","site_admin":false},"created_at":"2016-03-08T06:53:13Z","updated_at":"2016-03-08T06:53:13Z","body":"It's not clear as to what would happen in the following split brain scenario (scenario-1):\r\n1) split occurs, forming two networks\r\n2) the network that didn't have a master, elects a master (call this network-2)\r\n3) the master will elect a new primary (in network-2)\r\n4) 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)\r\n5) The connection between the networks is re-established.\r\n\r\nIn 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.\r\n\r\nAn example strategy could be:\r\n1) 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\r\nIf 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 created from network-1.\r\n\r\nAnother interesting situation (scenario-2) to consider:\r\n1) Continuing with the scenario described above until (4) ...\r\n2) 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.\r\n4) 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.\r\n\r\nThis would happen if there is no reconciliation strategy in effect.\r\n\r\nI 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.\r\n\r\nI am still trying to understand Elasticsearch's cluster behavior. It's possible that I might have made assumptions that aren't correct.\r\n"}}