[Discuss] Writes on NRT Replica with Remote Translog

[Bukhtawar]

Is your feature request related to a problem? Please describe.
For a remote translog in NRTReplicationEngine we ideally don't need to forward requests to the replica since the translog has already been durably persisted to a remote store. However in the event of a network partition where the primary and replica gets partitioned off, during this period, the primary might continue to ack writes and the replica(now promoted to a new primary) also does the same. This might continue longer espl with the #3045 even when there is no active master. As a result we can run into issues where we can have divergent writes and data integrity issues.
To avoid this we need to have a proposal to mitigate such risk
Also see #3237

Describe the solution you'd like

1. We do a no-op replication to all replica copies, this is similar to what we do today except for the No Op
2. We could have a mechanism sitting in the remote store to ensure both primary and replica reconcile their views based on a common agreed upon protocol based on versioning. If replica gets promoted it bumps up the version which the stale primary should be able to process and fail to ack subsequent writes.

Describe alternatives you've considered
A clear and concise description of any alternative solutions or features you've considered.

Additional context
Add any other context or screenshots about the feature request here.

[ashking94]

Well, I was thinking about this problem and realised that this would exist with segment uploads as well. Below is an use case -

1. Writer 1 creates segments and issues a write to storage. Filename - f1- contents C1
2. Writer 1 dies.
3. Writer 2 wakes up and issues a write to storage. Filename - f1- contents c2
4. With no OCC, we really can’t ensure that c1 does not override c2. Even if the file names are different, we will need a meta location that defines the segment location or the segment to writer mapping. The writes for that metadata should be conditional to ensure consistency.
   LMK your thoughts.
   cc @itiyamas

[ashking94]

The above scenario has been detailed and elaborated further in #3906 by @sachinpkale.

[ashking94]

Continuing further on the original point of discussion - Avoiding divergent writes with segment replication & remote translog

1. Context

Along with segment replication, we plan to have only one source of truth for storing the translog in remote store. Currently, in NRT segment replication mode, the segments are replicated asynchronously and the primary is responsible for forwarding the writes requests to replicas where the same request is replayed on primary without queuing the request in indexing buffer but only keeping track of the translog. Having translog in replicas help to have durability in check and as well as for the shard to validate if it is still the primary. On segments, only primary is responsible for its creation and it also notifies the replicas to sync the segment by using NRT checkpoints.

With remote translog on top of NRT segment replication, the remote store becomes the single source of truth for translog where just the current primary will be uploading the translog files. This obsoletes 1) the need for generation of translog on replicas as they can easily be replayed from the remote store in case of failovers/failures and 2) storing the translog on local disk on the primary. However, this does not necessarily eliminates the need for replication as the replication call also helps in validation for the assuming shard of being the real primary or not.

Now, if we decide to use remote store for translog and also stop replication requests, we get into this situation where due to network partition, the older primary can continue to accept write requests and cause divergent writes. The issue is further accentuated due to changes done to allow writes in #3621.

Invariants

• Durability - An acknowledged request will always be honoured in case of machine failures, failovers, recovery.
• Correctness & Integrity - Although durability guarantees that the data does not get lost once acknowledged. However, When the replica promotion happens, all the translogs must get replayed and should be correct in terms of ordering.

As mentioned in the current issue description, there are 2 possible options that are being discussed in further detail in following sections -

1. No-op replication - This is similar to replication, just that this would be no op. This way the partitioned primary can realise that it is no more a primary whenever the primary term changes. There is still a concern that must be addressed while dealing with indices with no replicas - more details in section 4. Other areas of concern.
2. Using remote store that guarantees read-after-write consistency (like s3) - We will use a remote store for storing the primary term which can help decide which is the active primary and which is the stale primary.

More details on the above 2 possible options follows below -

2. No-op replication

Currently if there are replicas configured, write requests are replicated synchronously across the insync replicas. This offers durability if the primary shard copy fails. When this happens, master leader promotes one of the insync replicas to become primary. If there are no insync replica, then index would show with red status. Now, with remote store for translog along with segment replication, we don’t need the translogs to be stored locally on the primary as well as the replicas. So, technically we can skip this part of keeping the write request in the indexing buffer and also skip the sync/async persistence of translogs on the replica’s data volume. This would require additional changes as mentioned in #3786. We can continue to use primary term validation logic that helps the stale partitioned primary to realise that it is no more a primary.

2.1 Concerns

• If there were no replicas, the stale primary would keep accepting write requests given we have Changing default no_master_block from write to metadata_write #3621. If auto restore kicks in, both stale and new primary can continue to accept requests because the connection between the stale primary and master (or other nodes) is no more. Stale primary would not know when to stop accepting requests. How do we make sure that the stale primary has stopped accepting the write requests?
  • We can tweak changes done in Changing default no_master_block from write to metadata_write #3621 for stale primary to stop requesting write requests. Auto restore can kick in then. This will make sure that, when the restore happens, the stale primary has failed totally.

3. Using remote store that guarantees read-after-write consistency

The primary term validation that we want to achieve in the above section 2. No-op replication strategy can also be modelled using a remote store. The following could be a strategy that the primary can follow -

• The flow whenever a replica is promoted to primary would look like below -
  • It reads the CurrentPrimaryTerm from remote store and validates the new primary term to be published is greater than the current primary term. Leader/Primary term metadata prefix for remote store: Base64UUID/IndexUUID/ShardID with filename as CurrentPrimaryTerm.
  • It publishes the new primary term to the remote store.
  • It starts downloading translogs basis the NRT checkpoint of the replica as segment replication would have copied over the segment files upto certain NRT checkpoint.
  • There might be a need to copy over the tanslog files from the previous primary term path to the new current primary term path in case of primary failing before the segment creation happens on the new primary. But this would depend on how translog replaying happens. I need to read more on this. TODO.
• The flow during an write request would look like below -
  • Primary receives the write request.
  • It stores the request in the indexing buffer and uploads the translog to the remote store. Translog prefix for remote store: Base64UUID/IndexUUID/ShardID/PrimaryTerm.
  • It reads the CurrentPrimaryTerm and validates against the local primary term.
  • If the local primary term and the CurrentPrimaryTerm don’t match, we respond with failure back to the client. However, since the data has been persisted and the new primary could have read the unacknowledged translog, we can allow that to happen and live without the need for a rollback.
    • There is a possibility that unacknowledged or failed write requests can get persisted. For such cases, we can use the last modified time of the translog file to determine whether to keep or discard it for replaying the translogs.

3.1 Concerns

• The remote store can become a single point of failure for all indexing operations and replica promotion. If the remote store is down, there can repetitive failovers or primary shard. There will be a need to devise a strategy on how the cluster behaves when the remote store is down irrespective of which strategy is selected to avoid divergent writes.

4. Other areas of concern

• With just segment replication & changes to allow write in absence of master Changing default no_master_block from write to metadata_write #3621, primary in partitioned node will continue to accept writes until it gets to know that it is no more primary from the replicas. In the case of no replicas, the partitioned primary will continue to accept writes indefinitely inspite of master not reachable. This behaviour is due to the change as mentioned in the #.
• This needs to be validated with code - During peer recovery handoff, can there be a case where a replica can get marked in sync and the old primary primary gets partitioned off just before the cluster state is published. Could this lead to a scenario where old primary and new primary can continue to act as primary in their individual capacity.

5. Considerations

• Translog durability guarantees
  • Like today, we can continue to use a setting similar to index.translog.durability which user can decide how frequently are the translogs persisted to the remote store. Both approaches provides this but at the cost of performance, correctness.
• Performance
  • Transport calls would be cheaper in terms of latency in comparison to remote store calls (which checks the latest primary term).
  • Remote store latency, throughput and throttling limits should be in check so that the remote store does not become the weakest link in the system.
• Plugin Extensibility (of primary term validation check)
  • No-op replication can continue to work irrespective of choice of remote store for segment & translog.
  • Remote store should be able to handle the abstraction of storing primary term metadata.
• Correctness
  • Since replication has already been the backbone for avoiding divergents, No-op replication approach can relied upon.
  • For remote store approach, the flow discussed above will work as long as the underlying remote store offers read-after-write consistency.
• Security
  • No changes in No-op replication approach.
  • OS would require CRUD access on the S3 bucket with the appropriate prefix.
• Operational Work
  • Ease of implementation & management
    • No-op replication approach requires minimal work on making the existing engine to have no op replication for replicas. It will require changes to be done in Internal Engine (for primary) such that concepts like global/local checkpoint can be refined. There will be no extra management overhead than it exists today with replication.
    • Remote store approach will also require changes in both the Engine that runs on the primary and the replicas. It would have similar effort around the changes in Internal Engine (for primary) as for No-op replication mode. The Engine on replica would be such that there should not be any calls that lands on it - be it index()/delete(). It exists only to be promoted to primary when the master asks it to become one. The abstraction would be built such that there are methods that persists/fetches the latest primary term, but there will onus on Remote store plugin implementors to extend and method as per the abstraction.
  • Complexity
    • Both approaches have the same kind of complexity.

5.1 Evaluation of approaches simplified

Considerations(↓) \ Approaches (→)	No Op Replication	Remote Store
Correctness	High	Dependent on underlying remote store
Performance	High	Low to Medium
Plugin Extensibility	Works for any choice of remote store for segments / translog	Requires abstraction of storing primary term metadata per Index Shard
Security	Status quo	OS requires CRUD on remote store for appropriate directory/paths/keys
Operational - Ease of implementation & management	Low to Medium	Medium to High
Operational - Complexity	Medium	Medium

Basis the evaluation, both the approaches can be used to handle divergent writes for remote translog. However, the major invariant i.e. durability and correctness, performance, and other considerations, we can modify current replication to make it no op and achieve the uber goal.

Please feel free to correct if I have stated any facts incorrectly and feedback/suggestions are welcome.

6. References

• Allowing writes in absence of master
  • PR - Changing default no_master_block from write to metadata_write #3621
  • Issue - Change default no_master_block from write to metadata_write #3045
• Handling replication for remote translog in primary and replica
  • Issue - [Remote store] Replication handling for remote translog for primary & replicas #3786

[ashking94]

cc: @mch2 @sachinpkale @Bukhtawar @kartg @andrross @reta @nknize. Looking for thoughts and suggestions.

[andrross]

Caveat: I'm certainly not an expert on this part of OpenSearch so please correct me wherever I'm wrong

One nitpick that might help clarify things is to replace the phrase "no op replication" with something like "primary term validation". If I understand correctly, you wouldn't need to send the document over the wire and instead would just be asking the replicas "is the primary term x?".

One concern I have about that approach is that today when a document is replicated via the translog, the replicas are in effect performing two operations transactionally: 1) validating that the primary term hasn't changed, and 2) persisting the document. With the proposed change those operations will be split apart. The replicas will confirm the primary term hasn't changed and the remote store will persist the document. However, those operations will be split apart to different distributed components, which makes me a little nervous. Can you provide more details about how this will work? e.g. do you upload to remote store before or after confirming primary term from the replicas? What happens if the primary term changes in between those operations? etc

Having said that, it does seem like a more natural fit to have the remote translog provide the same conceptual functionality provided by replicas today (persist this document only if the primary term is still x). It obviously puts a stronger requirement on the actual remote store because it has to be more than just dumb durable storage.

[reta]

A few comments from my side.

We can tweak changes done in #3621 for stale primary to stop requesting write requests. Auto restore can kick in then. This will make sure that, when the restore happens, the stale primary has failed totally.

I think this is the right thing to do, even in case of primary with no replicas: when the connection with master is lost (split brain etc), it looks right for the primary to go into readonly mode and not accepts writes anymore.

Leader/Primary term metadata prefix for remote store: Base64UUID/IndexUUID/ShardID with filename as CurrentPrimaryTerm.

It looks clever to store the CurrentPrimaryTerm (under Base64UUID/IndexUUID/ShardID) and check it before every write operation (so the primary could detect if it is stale or not). Storing translog under Base64UUID/IndexUUID/ShardID/PrimaryTerm would eliminate overlaps of old & new primary trying to update the translog simultaneously. But in this case the old translogs have to be recycled somehow, right?

[Bukhtawar]

One concern I have about that approach is that today when a document is replicated via the translog, the replicas are in effect performing two operations transactionally: 1) validating that the primary term hasn't changed, and 2) persisting the document. With the proposed change those operations will be split apart. The replicas will confirm the primary term hasn't changed and the remote store will persist the document. However, those operations will be split apart to different distributed components, which makes me a little nervous. Can you provide more details about how this will work? e.g. do you upload to remote store before or after confirming primary term from the replicas? What happens if the primary term changes in between those operations? etc

Thanks @andrross for the observation, Here is how we intend to solve the above, but before that some context on how doc repl does it, below are the sequence of steps.

1. An isolated primary indexes operations to Lucene and writes to it's local translog
2. Once it forwards the request to the replicas it then realises, it's partitioned off
3. The isolated primary would fail to ack this request to the client however the operation has been committed to the translog and runs the risk of exposing these operations for reads.
4. Soon after the no_master_block would kick in and help mitigate by stopping to accept further reads.

What follows from above is,

1. It's possible to have dirty reads which are unacknowledged
2. No acknowledged writes should be lost.

So if we go with the same status quo, we could always write operations to the remote xlog, once writes complete but before we acknowledge the write we can check for the primary term.

Having said that, it does seem like a more natural fit to have the remote translog provide the same conceptual functionality provided by replicas today (persist this document only if the primary term is still x). It obviously puts a stronger requirement on the actual remote store because it has to be more than just dumb durable storage.

This part is something we should discuss more. The implementation would probably require some leasing protocol

1. Writer Node(W1) hosting the primary shard tries to acquire a lease for shard id(S1) in the primary term(P1) for (x) seconds
2. Once acquired upload the translog to the remote store and renew the lease subsequently
3. Now, lets say remote store gets throttled and the plugin client retries
4. Meanwhile replica gets promoted to primary and writer node(W2) in primary term(P2) acquires the lease for shard id(S1) and starts the upload
5. W1 and W2 at this point could be concurrently uploading, unless W1's upload gets interrupted (JVM halted) once W1 fails to renew lease

We can always go with the "Remote Store" based proposal, which still provides us the limited guarantees we need. The part I like about this approach is it provides a nice property to better guard against isolated primaries as long as the cluster can guarantees the primary term invariant(no two primaries at any point can have the same primary term) at all times.

The "No Op replication" approach inherently treats remote xlog store an extension of the "primary" translog storage instead of treating it as a separate distributed store in a bid to ensure that writes ever happens from the primary of the shard instead of guaranteeing "a single writer" at all times.

The caveat here is that if the writer version (in this case primary term)change itself isn't guaranteed to be serialised, the remote store itself wouldn't be able to guarantee correctness, unless it implements some form of a leasing protocol mentioned above.

One thing to note here is that, the current snapshot mechanism(only primaries snapshot) also use remote store while keeping the remote store logic dumb. I think we should confirm how cases like isolated primaries are dealt there and whether remote store or remote xlog and can leverage some properties. It might be possible that snapshots can have certain trade-offs which which might not be acceptable for the current use case.

It looks clever to store the CurrentPrimaryTerm (under Base64UUID/IndexUUID/ShardID) and check it before every write operation (so the primary could detect if it is stale or not). Storing translog under Base64UUID/IndexUUID/ShardID/PrimaryTerm would eliminate overlaps of old & new primary trying to update the translog simultaneously. But in this case the old translogs have to be recycled somehow, right?

@reta Remote translog pruning would be discussed separately #3766

[Bukhtawar]

/cc: @itiyamas @muralikpbhat @backslasht