[RFC] Indexing and Search Separation

[amberzsy]

Is your feature request related to a problem? Please describe

Background
Currently, a data node performs both indexing and searching, leading to workload interference between these tasks. An expensive query can monopolize memory and CPU resources, causing indexing requests to fail or vice versa. Additionally, scaling read traffic typically involves adding more replicas, which can slow down indexing and reduce throughput. Therefore, supporting the separation of indexing and search will enhance read and write performance. Also Separation of indexing and search allows each to scale independently. For example, additional resources can be allocated to indexing processes during data ingestion, while search operations can be scaled separately to handle query loads.

Describe the solution you'd like

High level, there would be two approaches to achieve indexing and search separation. Node/Role level and instance/cluster level.

Node/Role level separation
In order to achieve Indexing and Search separation, we would build on the new node role “search” which separates out with existing data role which focus on indexing only. The “search” node role would act as dedicated search nodes.

With remote storage, we would keep committed data as segments and uncommitted data being added to the translog. To maintain consistency, the same semantics are applied when storing data in the remote store. Data from the local translog is backed up to the remote translog store with each indexing operation. Additionally, whenever new segments are created during refresh, flush, or merge processes, these new segments are uploaded to the remote segment store.

The control plane is running as active-standby for redundancy and would have built-in auto failover mechanism.
The search node directly downloads indexed data from remote storage and executes search operations, including aggregations. It operates in active-active mode to ensure availability during failures. The refresh interval should be configurable according to system limits.

Requirement:

1. traffic separation: coordinator role separation or consider have proxy layer atop for handling routing.
2. Cluster status - Green/Yellow/Red:
   a. today, any copy of shard in unassigned would trigger cluster status change to red(primary) or yellow. With
   separation, we would have higher granularity for indexing and search status. e.g when primary fails on serving write
   traffic, it would have certain indicator on Indexing Unhealthy/Unavailable. Similar apply to search, if any / all replica
   fails, should indicate search failure etc.
3. ShardAllocation strategy and zone/rack awareness:
   a. today, opensearch follows set of resiliency policy and allocation preference based on primary/replica architecture.
   With separation, the shard allocation would be based on role (search/data). For the awareness, primary shard would
   need to apply to primary active / standby (avoid allocate primary active and standby in one zone/rack). For search,
   try to distribute across different zone/rack.
4. Auto failover mechanism for primary active and standby. When primary active fails to serve traffic, it should
   automatically failover to primary standby for durability of indexing.
5. consistency guarantee:
   a. ensure consistency guarantees comparable to today's standards by having a search replica shard monitor its
   synchronization with the data node. If the replica is out of date, it can redirect or fallback the request to the data
   node. Options could include requiring strict consistency, allowing a maximum lag of X, and so on.
6. lightweight snapshot

Related component

Other

Describe alternatives you've considered

Cluster / domain separation
Alternatively, the similar indexing and search separation can be achieved through Cross-Cluster Replication with segment replication. With ccr-(segrep), the leader cluster will mostly handle the indexing/writes while the follower cluster will keep in-sync with segment replication. All indexing requests would route to the primary cluster and the search request route to the follower cluster.

Comparison:

Additional context

No response

[Pallavi-AWS]

This RFC refreshes some of the concepts discussed in an old RFC #7258 with more targeted preference around role separation. We'll review for overlaps and come up with an execution plan

[sohami]

I will share initial draft on how we can do some PoC with current available mechanisms to simulate reader/writer separation and experiments to show where this could benefit.

[mch2]

A really quick way to get node separation for experiments (without any standby writers):

1. Set preference in OperationRouting.getShards to Preference.REPLICA. This will route search requests away from the primary.
2. The TargetPoolAllocationDecider introduced for searchable snapshots allows us to make decisions based on a RoutingPool. We can update this to consider any replica as "Remote Capable" and force it to a node with the search role.

Played with this with a few test indices and ./gradlew run -PnumNodes=3 - code here

% curl localhost:9200/_cat/shards                                                         
test4 0 p STARTED 5 5.4kb 127.0.0.1 runTask-0
test4 0 r STARTED 5 5.4kb 127.0.0.1 runTask-1
test2 0 p STARTED 5 5.4kb 127.0.0.1 runTask-2
test2 0 r STARTED 5 5.5kb 127.0.0.1 runTask-1
test3 0 p STARTED 5 5.4kb 127.0.0.1 runTask-2
test3 0 r STARTED 5 5.4kb 127.0.0.1 runTask-1
test  0 p STARTED 5 9.6kb 127.0.0.1 runTask-0
test  0 r STARTED 5 5.5kb 127.0.0.1 runTask-1
% curl localhost:9200/_cat/nodes                                                          
127.0.0.1 23 99 15 2.33   dimr cluster_manager,data,ingest,remote_cluster_client * runTask-2
127.0.0.1 62 99 15 2.33   dimr cluster_manager,data,ingest,remote_cluster_client - runTask-0
127.0.0.1 60 99 15 2.33   s    search                                            - runTask-1

[andrross]

Just to build on a bit on what @sohami mentioned, here are the existing mechanisms I believe are relevant here:

• Remote-backed storage: With this feature, segment and translog durability is offloaded to the remote repository. Replica nodes will pull the segments from the remote store as opposed to relying on node-to-node replication.
• Search routing preference: A user can specific that searches are routed to replica shards. If all searches supply such a preference, then all search traffic would go to replicas (and implicitly all writes go to the primaries because only primaries can accept writes).

@sohami What else am I missing?

However, even with these mechanism we're still missing some pieces that would be needed for true index/search separation:

• By default, OpenSearch will attempt to evenly balance replica and primary shards across all nodes in the cluster. This is important in a homogeneous cluster to ensure the indexing load is spread evenly. For index/search separation we'd want exactly the opposite so that indexers and searchers are each allocated to a distinct subset of nodes.
• While remote-backed storage got rid of the need to replicate the translog across primaries and replicas because the remote store provides durability, it still kept the concept of "no-op replication" (discussed here) in order to protect against isolated primaries continuing to accept writes. With true indexer/searcher separation we probably would not want search replicas to be involved on the write path at all to provide true independent scalability.
• Somewhat related to the point above, we'd probably want the concept of this "standby primary" shown in the diagram to continue to provide the same consistency guarantees with no-op replication. OpenSearch today has only the "primary" and "replica" concepts so this would likely be a new concept. Alternatively, if the remote store could provide consistency guarantees (this would likely require a different type of store than an object store like S3), and OpenSearch could quickly provision a new primary fast enough in the case of a node failure, then the standby might not be necessary.

[sohami]

@andrross and @mch2 Yes I am thinking on similar lines and to @andrross point about separating indexers and searchers, we can use an existing index setting for PoC.

With true indexer/searcher separation we probably would not want search replicas to be involved on the write path at all to provide true independent scalability.

For now I was thinking we can keep this as is because with remote store the message between indexer and search will be of light weight (hence also calling out in Assumption below that we plan to mainly support for remote store based indices). Majority of the work will be involved in downloading the data from remote store which searcher anyways will need to do in separated setup as well.

This is what I think could be a good starting point for the PoC, let me know your thoughts.

Benefits of Reader/Writer Separation:

• Failure isolation between readers and writers
• Independent scaling of ingestion and search workloads as needed
• Independent workload based optimizations for reader and writer:
• a) Using available knobs to tune
• b) Selecting different instance/node types. This can potentially help to save on the cost depending on different instance type needed for indexing/search workloads. For e.g. compute optimized instances are cheaper than memory optimized instances

Instance Type	vCPU	Memory	Storage	Price per hour	Type
c6g.large.search	2	4	EBS Only	$0.113	Compute Optimized
r6g.large.search	2	16	EBS Only	$0.167	Memory Optimized

Assumptions:

True reader and writer separation will make sense with segment replication and remote store, so we will not consider doc rep based indices in the mix or segrep indices with local store.

Suggested PoC:

To achieve the reader/writer separation we can use the follow setup.

• Create a cluster with
• 2 coordinator nodes, this can be achieved by setting up nodes with no role
• 2 nodes with data role (use CPU optimized c6g.large ones to make it easy for repro) and
• 3 dedicated cluster manager nodes
• Create an index with 1 shard having 1 primary and 1 replica
• Update index setting index.requests.cache.enable to disable request cache
• Optional: Apply the index setting index.routing.allocation.total_shards_per_node to 1 to keep 1 shard per node. This will be needed if we plan to increase the node count and replica count. Otherwise in 2 node setup without having this setting is also fine.
• OSB client should communicate to the coordinator only nodes
• We may need to modify OSB to send search requests with a preference parameter in the request (if not already supported). Ref: https://opensearch.org/docs/latest/api-reference/search/#the-preference-query-parameter
• Other option is to create an index setting and just force use that preference which is set in the index setting to figure out the shard copy. Ref code pointer: https://github.com/opensearch-project/OpenSearch/blob/main/server/src/main/java/org/opensearch/cluster/routing/OperationRouting.java#L244

PoC Tests:

Scenario 1:

Search traffic on any shard (primary or replica) on a node can affect the indexing traffic on same node to either same or different index shard.

1. Start the indexing traffic on the test index. The indexing load will go to the primary shard and replica will try to sync from remote store
2. Start the search traffic on the index which is memory intensive and is targeted only to the primary shard using the request level preference parameter set to _primary. This will simulate the behavior that same index primary shard is serving both read/write traffic.
1. We are using this setup to keep it simple to simulate otherwise we can do the same thing with shards of different index such that on a node primary of 1 index and replica shard of other index is colocated.
2. For memory intensive query, we can try the multi-terms aggs query or use the example here: [BUG] A sufficiently small interval value on a histogram can crash the node #14558 (comment)
3. We can do 2 flavors here,
1. Increase the search traffic and keep the indexing traffic constant until the node drops out of the cluster due to overload.
2. Increase the indexing traffic and keep the search traffic constant until the node drops out of the cluster due to overload
4. Once we have the workload which can trigger search/indexing overload, we can see the impact on other request types i.e. search affecting indexing (a) or vice-versa (b)
5. Now using the same workload as above, we can re-run it but directing the search traffic to replica shards using preference=_replica
6. In step 5, we should not encounter any failure which can showcase the workload isolation for search and indexing traffic.
7. Now keep increasing the search traffic on the replica shard, until the node hosting replica shard drops out of the cluster. Once this happens there should not be any impact to the indexing traffic. This will showcase the failure isolation for indexing and search
1. Note: We cannot do the same thing for indexing failure isolation as that will trigger primary promotion and cause search traffic failure as well.
8. Now add another node and increase the replica count to 2. Add the index setting of total_shards_per_node of 1 to the index. Again trigger the same indexing and search traffic as in step 7 which caused the replica node crash. Now with additional replica, we should be able to handle the search traffic. This will showcase the search scaling independent of indexing.

Scenario 2:

Different instance types for indexing and search. In shared setup, if we need to scale to memory optimized instance then both nodes hosting primary (writer) and replica (reader) shards will need to scaled up whereas in this case only node hosting replica (reader) shard is scaled up showing the cost benefits.

1. With above setup and difference being to use CPU optimized instances (c6g.large) for primary and replica nodes and performing workload in step 7.
2. Now update the replica node instance type to Memory optimized instance (r6g.large) instead of CPU optimized one. Again perform the workload in step 7, we should see replica able to handle the workload without dropping from cluster. This will showcase that indexing and search can be optimized with the instance type suitable for each workload types.
1. We may need to manually reroute the shards to ensure primary and replica is allocated on correct instance types.

[andrross]

For now I was thinking we can keep this as is because with remote store the message between indexer and search will be of light weight

@sohami Can you show the failure isolation aspect with no-op replication in place? If searches brown-out a replica, I think that will impact the indexer because it will still be waiting on acks from the replica.

showing the cost benefits

I think it would be interesting to compare the performance of two clusters: one with specialized nodes of different types, and one with all nodes of the same type and roles, but in both cases the total cost of all the nodes is the same. Then it would be interesting to see in what scenarios the specialized nodes can give better performance.

[sohami]

For now I was thinking we can keep this as is because with remote store the message between indexer and search will be of light weight

@sohami Can you show the failure isolation aspect with no-op replication in place? If searches brown-out a replica, I think that will impact the indexer because it will still be waiting on acks from the replica.

Good point. My thought was with remote translog probably the no-op replication will not affect indexing but seems like it does (thanks to @mch2 for confirming). Then we will need to bypass that for PoC and make it a "dummy" one to always be successful.

showing the cost benefits

I think it would be interesting to compare the performance of two clusters: one with specialized nodes of different types, and one with all nodes of the same type and roles, but in both cases the total cost of all the nodes is the same. Then it would be interesting to see in what scenarios the specialized nodes can give better performance.

Need to think more on this. Let me know if you have any suggestion on how to achieve this. At first thought, I think we probably need some workload such that indexing and search are competing on one of the resource (CPU/memory) which can be solved by adding a new instance type for indexing/search (this is what the suggested experiment scenario 2 is showing). But doing that with similar cost between 2 setup could be challenging specially when there is not significant difference between instance types. So I think it will require a bigger setup probably with large instance count. Creating a specific targeted workload at such big setup will then become a challenge.

[amberzsy]

@sohami Can you show the failure isolation aspect with no-op replication in place? If searches brown-out a replica, I think that will impact the indexer because it will still be waiting on acks from the replica.

Good point. My thought was with remote translog probably the no-op replication will not affect indexing but seems like it does (thanks to @mch2 for confirming). Then we will need to bypass that for PoC and make it a "dummy" one to always be successful.

curious on why indexer would wait on acks from the replica (or brown-out replica can impact indexer) as per my understanding, it would ack on doc being flushed to translog in storage for segrep with remote storage.

[andrross]

curious on why indexer would wait on acks from the replica (or brown-out replica can impact indexer)

@amberzsy You can find the full context here: #3706

[yupeng9]

Curious, do we need to upload the translog to the remote storage or just the committed segments? I think the data nodes for indexing still need to have node-to-node replication for redundancy, but the search nodes need to download the committed segments only for serving?