[RFC] Cloud-native OpenSearch

[yupeng9]

Is your feature request related to a problem? Please describe

Elasticsearch/OpenSearch, launched in 2010, was initially designed for on-premise deployments and smaller-scale applications, when cloud-computing was not prevailing at that time. This was shown in these observed design patterns:

• Bundled Roles: Elasticsearch combined multiple roles (data storage, search, master node responsibilities) within a single node. This simplified setup and management for single-node or small clusters.
• Mesh Design for Metadata: Cluster metadata was propagated to every node using a mesh design. This provided resilience against node failures as any node could provide the necessary information.
• Data Movement Algorithms: Elasticsearch included algorithms to move data around the cluster. This was essential for rebalancing data in response to cluster growth, node additions, or node removals.

While this design initially provided convenience and ease of use, it has encountered significant challenges as cloud computing has become dominant and use cases have grown in scale and complexity:

• Reliability Challenges at Scale: Managing very large Elasticsearch clusters (e.g., exceeding 200 nodes) becomes increasingly difficult. Data movement operations and cluster metadata replication can be time-consuming and lead to cluster pause.
• Rolling Restarts and Upgrades: Performing rolling restarts or upgrades on large clusters can introduce latency spikes and potential data inconsistencies. Rollbacks can also be problematic. While Blue/Green deployments offer a safer alternative, they require significant additional capacity, which can be costly.
• Challenges with Reactive Scaling: Dynamically adding or removing replicas in response to changes in serving traffic is challenging. Additionally, imbalances in traffic across shards can lead to uneven replica distribution, impacting performance and resource utilization.
• Lack of Shard Distribution Control: Managing shard distribution in large clusters is complex. Fine-tuning shard size and distribution can be crucial for optimizing performance, especially for latency-sensitive applications.

We propose transitioning OpenSearch to a cloud-native architecture. In the cloud era, the shift towards cloud computing and the demands of modern, large-scale applications necessitate a rethinking of OpenSearch's architecture. A cloud-native architecture, leveraging cloud services and adopting a shared-nothing design, can address many of these challenges:

• Cloud Components: Utilizing cloud-based services for storage, compute, and networking can provide greater scalability, elasticity, and resilience.
• Shared-Nothing Architecture: In a shared-nothing architecture, nodes operate independently and do not share resources. This can simplify cluster management, improve scalability, and reduce dependencies between nodes.

By embracing a cloud-native approach, OpenSearch can better meet the requirements of large-scale, cloud-based deployments. This transition can lead to improved operational efficiency, enhanced reliability, and greater scalability, enabling OpenSearch to handle the demands of modern data-intensive applications.

Describe the solution you'd like

Goal

Remove the 200-node limit on cluster size to enable high scalability.
Allow rolling upgrades and restarts with a 20% capacity buffer, eliminating the need for Blue/Green deployments.
Support rapid horizontal scaling per shard to provide elasticity.
Improve the reliability and reduce the operational work for large clusters
Allow shard size and distribution fine tuning for better query performance

Architecture

The proposed architecture is shown as in Figure 1. And it includes several modular changes compared to the architecture today.

Figure 1. Cloud-native OpenSearch architecture

External Metadata Storage

An important architectural change that can be made is the decoupling of logical metadata and physical cluster state, and storing the metadata in a shared external storage system. This would result in better scalability and faster access. Currently, cluster state propagation to all nodes can be slow in large clusters.

This idea was also proposed in an RFC that lists faster index creation/updates, better memory efficiency, and faster crash recovery among the benefits of this change. An additional benefit to external metadata storage is that it can simplify control plane implementation and delegate durable metadata state management to an external system, such as etcd/zookeeper, which are known to be reliable, distributed coordination systems. The RFC recommends RocksDB for metadata storage; however, RocksDB is not a distributed system and therefore does not provide the high availability and durability of strongly consistent KV storage systems such as etcd/zookeeper. Many technologies, including Kubernetes and Apache Kafka/Pinot/HDFS/Yarn, utilize etcd and zookeeper for metadata storage and coordination. As the Hadoop ecosystem is being replaced by more modern, cloud-native, and real-time data tools, it is proposed that etcd be the default external metadata storage compared to zookeeper. The metadata storage layer can be abstracted to allow for other storage options, such as zookeeper.

Additionally, etcd provides features such as watchers that are valuable for nodes in building reactive, event-driven features, including leader election, service discovery, and dynamic configuration.

Shared-nothing Data Nodes

Another important architecture change to enable cloud-native design is to make the data nodes shared-nothing. This means that each data node is independent and does not need to communicate with other data nodes. As shown in Figure 1, this is possible by the following changes:

• Use segment-replication instead of the document-replication, so that the index segment is fetched from the remote storage (e.g. S3)
• Use the external metadata storage as described above as the logical data and its goal state, so that a newly data node knows which index and shards that it serves along with other configurations. Unlike today’s design, the data node needs to fetch only the metadata relevant to it but not the global metadata. Also, the data node does not need to communicate with the master node to collect the cluster state.
• Use a dedicated coordinator for serving the queries, so that the data node does not need to maintain the global cluster state for routing and forwarding the queries to other data nodes like today’s design
• A pull-based ingestion (introduced in this RFC) can further enhance the shared-nothing design as it removes the need of translog, so a data node can quickly recover by replaying the messages from the streaming source. In fact, with the pull-based ingestion, we can even enable the leaderless (or primary-less) design because all the shard replicas are equivalent to the primary in terms of data ingestion capability.

This shared-nothing design brings in many benefits:

• Horizontal scalability, the data nodes can be scaled out linearly and quickly without any shared bottlenecks. This is valuable for auto-scaling based on the traffic pattern.
• Reduced contention, since the cluster state is not shared among data nodes, there is no waiting on its propagation
• Better fault isolation. Since a data node does not depend on others, one data node crashing does not affect the other nodes, which improves the cluster resilience.
• Easier operation. Since the data nodes become more modular, they are easier to operate and maintain. We can add/remove data nodes similar to microservices, and perform rolling restart/upgrade without the 2X capacity needed for Blue/Green deployment.

Controller with Goal State vs Actual State Convergence Model

With the metadata stored in an external storage system, we could simplify the master node design to adopt the goal state vs actual state convergence model, a design pattern proven to be resilient in large scale distributed systems with real-world examples like Kubernetes. A system that naturally drifts toward correctness, can provide strong self-healing capabilities and thus ease the operation and maintenance.

With the metadata stored in the external storage, the Controller can also store the goal state of the data nodes, such as the index and shard to serve for a given data node. The data node also writes its actual node state in the external storage such as the node health, configurations of the index it is serving. The controller can watch and compare the goal state and actual state, and update the goal state when needed, such as the leader selection of a shard to a given data node.

Stateless Coordinator

With the external metadata storage, it’s possible to make the coordinator a stateless service, so that it does not need to join the cluster like the data nodes . It can watch the metadata store and therefore it does not need to receive the cluster state from the master node. The cluster state includes the actual state of data nodes such as the shard allocation and node health, which can be used for building the routing table for the coordinator to dispatch the queries. A stateless coordinator can greatly simplify the deployment/operation, and it allows linear scalability of the coordinator horizontally.

Related component

Cluster Manager

Describe alternatives you've considered

No response

Additional context

This RFC is also written in this google doc for comments and discussion

[guojialiang92]

Hi, @yupeng9, I think it's a great idea and would like to discuss it.

External Metadata Storage
An important architectural change that can be made is the decoupling of logical metadata and physical cluster state, and storing the metadata in a shared external storage system. This would result in better scalability and faster access. Currently, cluster state propagation to all nodes can be slow in large clusters.

Will data nodes frequently access external Metadata storage (etcd/zookeeper) under the Cloud-Native architecture? Compared to the current Metadata stored locally on the node, I am concerned that frequent access to external metadata through the network will have a performance impact. (Compared to k8s, online search engines may face more metadata updates?)

[yupeng9]

Hi, @yupeng9, I think it's a great idea and would like to discuss it.

External Metadata Storage
An important architectural change that can be made is the decoupling of logical metadata and physical cluster state, and storing the metadata in a shared external storage system. This would result in better scalability and faster access. Currently, cluster state propagation to all nodes can be slow in large clusters.

Will data nodes frequently access external Metadata storage (etcd/zookeeper) under the Cloud-Native architecture? Compared to the current Metadata stored locally on the node, I am concerned that frequent access to external metadata through the network will have a performance impact. (Compared to k8s, online search engines may face more metadata updates?)

This is a good question. Usually etcd allows thousands of clients to connect with 3k+ ops/sec for write and 20k+ ops/sec for read based on its use in kubernetes stack. This shall be sufficient to support a cluster of thousands of nodes based on the known industry use. Also, each data nodes can also cache metadata locally so that we can reduce the frequency of access and leverage watchers to be notified when changes are made. I'm also curious about kind of frequent metadata updates that you have in mind?

[navneet1v]

@yupeng9 Thanks for creating the proposal. If I am not wrong there has been several attempts being done in past to build a cloud native arch for OpenSearch(Ref: #13274 and another one you mentioned in your proposal, there can be more which I might be missing). This is a good proposal, but I want to understand what aer your thoughts on some of the core challenges for such arch:

1. The current metadata/cluster state being stored in OpenSearch is stored as 1 big blob in RAM of master nodes, does this new arch solves that problem? Because I think its pretty critical for scaling for large number of nodes.

In the proposal I see readers and writers of a shard on same node? So with this proposal we are not solving the reader/writer separation and shard placement problem?

etcd where cluster state is getting stored will that be the only implementation? or we will keep it open for extension?

This is a good question. Usually etcd allows thousands of clients to connect with 3k+ ops/sec for write and 20k+ ops/sec for read based on its use in kubernetes stack. This shall be sufficient to support a cluster of thousands of nodes based on the known industry use.

This ops/sec would be at a specific size of the response. With > 200 nodes and say thousands of indices with large index mappings, will we get same ops/sec?

[yupeng9]

1. The current metadata/cluster state being stored in OpenSearch is stored as 1 big blob in RAM of master nodes, does this new arch solves that problem? Because I think its pretty critical for scaling for large number of nodes.

Agreed. The separation metadata and cluster state is part of this, and we will work on a spec to list the state and also how we can separate them. so that we can have a more finer-granular control which piece of metadata needs to be stored/fetched.

In the proposal I see readers and writers of a shard on same node? So with this proposal we are not solving the reader/writer separation and shard placement problem?

Reader/Writer separation is an ongoing effort, so it's orthogonal to this effort, though reader/writer separation is a good pattern for cloud-native architecture.

etcd where cluster state is getting stored will that be the only implementation? or we will keep it open for extension?

It will be an API, and etcd-based implementation will be one plugin and the likely one that we'll build the prototype with.

This is a good question. Usually etcd allows thousands of clients to connect with 3k+ ops/sec for write and 20k+ ops/sec for read based on its use in kubernetes stack. This shall be sufficient to support a cluster of thousands of nodes based on the known industry use.

This ops/sec would be at a specific size of the response. With > 200 nodes and say thousands of indices with large index mappings, will we get same ops/sec?

Right, this ties to the first question. If we always store a gigantic blob cluster state, then the throughput will be limited. But if we can store things in finer granular, and update the key of the changed metadata only, I believe we can improve the throughput. wdyt?

[navneet1v]

Hi @yupeng9 thanks for providing the response. Please find my response below:

1. The current metadata/cluster state being stored in OpenSearch is stored as 1 big blob in RAM of master nodes, does this new arch solves that problem? Because I think its pretty critical for scaling for large number of nodes.

Agreed. The separation metadata and cluster state is part of this, and we will work on a spec to list the state and also how we can separate them. so that we can have a more finer-granular control which piece of metadata needs to be stored/fetched.

Super awesome. It will be good if we can mention this in proposal.

In the proposal I see readers and writers of a shard on same node? So with this proposal we are not solving the reader/writer separation and shard placement problem?

Reader/Writer separation is an ongoing effort, so it's orthogonal to this effort, though reader/writer separation is a good pattern for cloud-native architecture.

agreed.

etcd where cluster state is getting stored will that be the only implementation? or we will keep it open for extension?

It will be an API, and etcd-based implementation will be one plugin and the likely one that we'll build the prototype with.

Make sense. Love the idea of plugin will look to see more details on this.

This is a good question. Usually etcd allows thousands of clients to connect with 3k+ ops/sec for write and 20k+ ops/sec for read based on its use in kubernetes stack. This shall be sufficient to support a cluster of thousands of nodes based on the known industry use.

This ops/sec would be at a specific size of the response. With > 200 nodes and say thousands of indices with large index mappings, will we get same ops/sec?

Right, this ties to the first question. If we always store a gigantic blob cluster state, then the throughput will be limited. But if we can store things in finer granular, and update the key of the changed metadata only, I believe we can improve the throughput. wdyt?

If the metadata is split into smaller granular chunks then yes provided throughput should be achieved. Will be looking forward for the proposal on splitting the metadata into chunks.

I have more question/ask/thought and would like to know what you think about it:
with the proposal mentioned in this GH issue will be able to achieve say millions of indexes in a cluster? Because in my mind the problem was always with metadata and shard management. I think with this proposal it should be resolved. If we are not thinking about this, can we take this as requirement for the current proposal.

[guojialiang92]

This is a good question. Usually etcd allows thousands of clients to connect with 3k+ ops/sec for write and 20k+ ops/sec for read based on its use in kubernetes stack. This shall be sufficient to support a cluster of thousands of nodes based on the known industry use. Also, each data nodes can also cache metadata locally so that we can reduce the frequency of access and leverage watchers to be notified when changes are made. I'm also curious about kind of frequent metadata updates that you have in mind?

Thanks, @yupeng9. The access frequency may depend on the number of nodes/number of indices/mapping size. In this architecture, we may no longer need to split more clusters for stability, so millions of indexes may be attractive to me.

I still have some questions I want to confirm with you:

1. Is there still any form of communication between the master node and the data node in the Cloud Native architecture? In my understanding, it is not necessary. Under the new architecture, only the master node needs to upload the target state to the external Metadata storage, and the data node subscribes to the target state to the Metadata storage and reports the actual state.
2. Splitting a big blob of metadata into finer grains is very helpful for the scalability of the cluster (Based on your discussion with @navneet1v). Also, I don't think this job relies on external metadata storage. In other words, are these two jobs orthogonal?

[navneet1v]

2. Splitting a big blob of metadata into finer grains is very helpful for the scalability of the cluster (Based on your discussion with @navneet1v). Also, I don't think this job relies on external metadata storage. In other words, are these two jobs orthogonal?

@guojialiang92 your thought is right splitting the metadata and external metadata storage is independent, but splitting metadata into smaller chunks actually makes the external metadata storage work simpler and scalable. This also ensures that clusters with large number of nodes can be created(>200 nodes), which was one of the main motivation for this proposal.

[yupeng9]

Thanks, @yupeng9. The access frequency may depend on the number of nodes/number of indices/mapping size. In this architecture, we may no longer need to split more clusters for stability, so millions of indexes may be attractive to me.

Curious, millions of indexes is a lot, why do we need to support this much?

I still have some questions I want to confirm with you:

1. Is there still any form of communication between the master node and the data node in the Cloud Native architecture? In my understanding, it is not necessary. Under the new architecture, only the master node needs to upload the target state to the external Metadata storage, and the data node subscribes to the target state to the Metadata storage and reports the actual state.

Yes, that's my feeling too that master node needs to update the goal state, and it does not need to handle the state propagation.

3. Splitting a big blob of metadata into finer grains is very helpful for the scalability of the cluster (Based on your discussion with @navneet1v). Also, I don't think this job relies on external metadata storage. In other words, are these two jobs orthogonal?

Yeah, these two can be independent, but when we design the content of the external metadata storage, we need a spec of the metadata layout.

[navneet1v]

Thanks, @yupeng9. The access frequency may depend on the number of nodes/number of indices/mapping size. In this architecture, we may no longer need to split more clusters for stability, so millions of indexes may be attractive to me.

Curious, millions of indexes is a lot, why do we need to support this much?

@yupeng9
I don't think so millions of indices is a lot. If you look at systems like Turbopuffer they support tens of millions of indices. https://turbopuffer.com/docs/limits . The main use case of millions of indices is multi-tenancy and tenant isolation. Idea is rather than creating 1 single giant index with all the tenants data and then search using filters, user can create indices per tenant and then search based on that. This is very clean, easy to scale and maintain from a user side, provides awesome query latency.

So I believe with this proposal we should be able to achieve millions of indices. I do agree we might need more work in other part of OpenSearch to make this work, but as per my deep-dive and understanding splitting metadata into smaller chunks is one of the requirement to support multiple indices, followed by not keeping shards on the node(writable warm) to reduce inode usages.

[yupeng9]

@yupeng9 I don't think so millions of indices is a lot. If you look at systems like Turbopuffer they support tens of millions of indices. https://turbopuffer.com/docs/limits . The main use case of millions of indices is multi-tenancy and tenant isolation. Idea is rather than creating 1 single giant index with all the tenants data and then search using filters, user can create indices per tenant and then search based on that. This is very clean, easy to scale and maintain from a user side, provides awesome query latency.

Yes, I agree multi-tenancy makes sense but millions of indices is still a lot.. Would that mean tens of thousands of customers sharding the same opensearch cluster? I feel the data size and cluster size will first run into the limit before the number of indices