[Segment Replication] Support shard promotion.

[mch2]

Support failover scenario of a replica re-assigned as a primary.

Right now this flow fails because of how we wire up the engine as readOnly by setting IndexWriter to null.

Issues remaining (updated 7/22):

• [Segment Replication] Update shard promotion algorithm to consider replica checkpoints. #3988
• [Segment Replication] Swap replica to writeable engine during failover. #3989
• [Segment Replication] Swap primary engine to NRTReplicationEngine during handoff. #4007
• [Segment Replication] Synchronize checkpoint updates with failover. #4136

[hydrogen666]

How you guys handle duplicate file name afer replica shard promition?
For example, we have a index with 1 shard 2 replica configuration, it has two replicas A and A', primary shard goes down,master choose one replica A to be promotted to primary shard, replica A contains segment _1, _2, _3 before promotion. But another replica A' is much newer, it contains segment _1, _2, _3, _4. During A promotion progree , it need to run tlog recovery progress, it will generate segment _4' which is not same as _4 in replica A'. How you guys handle this scenario?

[mch2]

Hi @hydrogen666 thanks for the question!

In the scenario you describe once the tlog recovery completes on A it would publish a new checkpoint notification to replicas. The published checkpoint data also includes a primaryTerm field. When this is incremented the replicas would process this checkpoint and fetch the new segments from the primary, even if the latest sequence numbers match. The copy process starts with replicas fetching a list of the latest segment metadata available the primary and then computing a diff against its own that includes different and missing files. In this case A' will recognize its version of _4.si is different and replace it with the primary's version.

With that said I haven't yet put together a failover section to be included in the proposal on what failover will look like with segrep. If you have suggestions on how you think this should work please share them, thanks!

[hydrogen666]

Hi @mch2
Yes, A' can recognize its _4.si is not same as the newly generated _4.si by new primary. But how to handle if someone's IndexSearcher on A' is still holding the origin _4.si ( scroll request will hold a segment for a long time) ?

I can only think of theses ways to deal with it:

1. Switch a new Directory and new SearcherManager after the replica found primaryTerm is bumped ?

2. Send a _{n} (n is larger than the largest segment counter in current SegmentInfos) to master node before segment replication progress so that master can tell the newly promoted Primary Shard not to generate any segment less than _{n}. To reduce the pressure of master node, we don't need to send this information every time. For example, if the max segment is _4.si in primary's current SegmentInfos, we can send _rw(or 1004 in decimal) to master node. After segment increases to _rw.si, we send _1jo(or 2004 in decimal) to master node.

3. Before choosing replica doing promotion progress, master node must ask all replicas whose replication state is newest.

[mikemccand]

These are awesome questions about segment replication! This is indeed a challenging situation for segrep, but is solvable with one of the three proposed options.

Lucene is fundamentally a write-once (at the OS filesystem layer) index, so we must ensure that no node (primary or replica that was just newly promoted as primary) ever attempts to overwrite an existing index file on any of the existing/prior replicas.

Is solution 3) too hard to do? I would think distributed consensus algorithms could do the right thing here -- all replicas that are currently reachable could respond with how far they had replicated and we pick the furthest one to promote to the new primary? Any detached (network partitioning) replicas that come online too late must possibly delete "future" segment files before re-joining the cluster, which should be fine since those detached replicas are not in use...

Also note that scan/scroll is a weak implementation in ES/OS today, with the hard routing back to a single replica which might drop offline -- with segrep, any replica can accurately/precisely service the scan/scroll request (since all replicas share precise point-in-time views of the shard), and hold allocated (i.e. cannot delete, keep an open IndexReader/Searcher until the lease expires) a consistent point-in-time snapshot from primary.

With segrep, every replica can search accurate point-in-time snapshots of the index, unlike ES with its inefficient document replication today where every replica is searching "slightly" different point-in-time views of the index. This can make for very accurate pagination, versus today where users paging through results might easily see duplicates on page N+1, or might also miss results, eroding trust.

Failing that (option 3 is too hard), I think option 2 is best -- bump the next segment number to a number clearly larger than any replica might have already copied. Lucene uses a long to increment these segment numbers so we have tremendous headroom to "jump ahead" with near zero risk of running out of the 64 bit space.

[nknize]

With segrep, every replica can search accurate point-in-time snapshots of the index

Noting here there's a Point in Time API design issue and corresponding PR to incrementally improve the scroll shortcomings in OpenSearch using PIT but it will most certainly need to be refactored under segrep.

Thinking out loud (haven't mulled this one as much) I like option 3 but it seems like a lot of coordination and we'll certainly want to benchmark? Option 2 might be an "easier" near term implementation and we can migrate to the coordinated approach in a follow on major release.

[andrross]

Also thinking out loud, I do kind of like option 2 as a good guarantee that we'll never reuse a segment number no matter what weird distributed system edge case is encountered. We could add the logic to choose a new primary based on which replica has made the most progress as an optimization while still keeping the segment high water mark logic. It would allow the "choose farthest ahead replica" logic to be best effort as opposed to being critical for correctness.

[mch2]

Thanks all for your thoughts here.

I like the idea of using both 2 and 3. I don't think implementing 3 will be all that hard because each replica already updates their local checkpoint and publishes an updated global checkpoint to cluster state after a copy event completes. I wonder if we could reuse that global checkpoint mechanism in reverse to fetch the furthest ahead replica.

With segrep, every replica can search accurate point-in-time snapshots of the index, unlike ES with its inefficient document replication today where every replica is searching "slightly" different point-in-time views of the index. This can make for very accurate pagination, versus today where users paging through results might easily see duplicates on page N+1, or might also miss results, eroding trust.

Thanks for calling this out. For this guarantee I'm assuming we will need to block primaries from opening a new reader until we have confirmed all replicas have received and opened up the latest copy? Our implementation isn't doing this right now and only initiates a copy event after the primary has refreshed. I wonder if this is something worth gating with an additional setting?

[anasalkouz]

Is segment number is the only factor to decide what replica to promote? Do we need to take shard allocations, node load ..etc in our consideration?

[mikemccand]

With segrep, every replica can search accurate point-in-time snapshots of the index, unlike ES with its inefficient document replication today where every replica is searching "slightly" different point-in-time views of the index. This can make for very accurate pagination, versus today where users paging through results might easily see duplicates on page N+1, or might also miss results, eroding trust.

Thanks for calling this out. For this guarantee I'm assuming we will need to block primaries from opening a new reader until we have confirmed all replicas have received and opened up the latest copy? Our implementation isn't doing this right now and only initiates a copy event after the primary has refreshed. I wonder if this is something worth gating with an additional setting?

Hmm no that should not be necessary?

You should instead use Lucene's SearcherLifetimeManager class to efficiently hold open any past point-in-time readers that are still leased by running scan/scroll requests. This way every replica is free to refresh after each checkpoint, while keeping the older readers still available. Each reader is keyed by a long version that increments on each checkpoint, and should be persisted into the scan/scroll state. Primary can refresh first, and handle a new incoming scan/scroll at the latest version. If that scan/scroll client comes back for page two, we will need some added (cluster state? not sure) logic/storage to record which of the replicas have that index version refreshed now. Those replicas would also need to know that there is a lease held on version N and keep that reader alive (e.g. using SearcherLifetimeManager).

[mikemccand]

Is segment number is the only factor to decide what replica to promote? Do we need to take shard allocations, node load ..etc in our consideration?

Do Primaries handle searching too? Or are they dedicated to indexing (which would give awesome physical isolation of JVMs that do searcher versus indexing, but would require more total cluster resources typically).

If Primaries do searching too then I agree we may want to take shard allocations / indexing rate across those shards into account. We have some freedom to promote a Replica on a relatively underutilized node, instead of a Replica on a node that is already doing too much indexing/searching? In fact, this should be a powerful freedom for balancing load across the cluster ... we may want to do it (promote a different Replica as Primary for this shard) pro-actively (not just during node failure cases) to keep the cluster load balanced?

[nknize]

use Lucene's SearcherLifetimeManager class to efficiently hold open any past point-in-time readers that are still leased by running scan/scroll requests

This sounds like a better mechanism to use within the SearchService#createPitReaderContext logic flow? Looks like the proposed PR leverages all of the inherited keep alive logic from the existing scroll API whereas SearcherLifetimeManager can take care of this for us? /cc @bharath-techie.

I think we should look at refactoring away from the old Scroll scaffolding and move toward using this mechanism in that PR?

[nknize]

Do Primaries handle searching too?

For now, yes. This was discussed for "priority queries" (e.g., alerting use cases) where the primary would have documents available for search before replicas.

Or are they dedicated to indexing.

This is the desired next step; to isolate the reader / writer jvms. At minimum I think we'll want to look at leveraging some of the concurrency improvements being made in lucene to help in the decoupling of indexing and search time geometries?

[nknize]

/cc @vigyasharma to join the discussion!