Skip to content

Commit df9b566

Browse files
synhershkosynhershko
committed
Add replica-only node type for read scalability
This commit introduces a new `replica_only` node role that provides read scalability without impacting the write path or triggering cluster rebalancing, enabling cost-effective horizontal scaling for read-heavy workloads. ## Motivation Organizations often need to scale read capacity independently from write capacity. Traditional approaches of adding more data nodes cause: - Unwanted shard rebalancing across the cluster - Increased write coordination overhead - Higher infrastructure costs for full-featured data nodes The replica-only node type addresses these challenges by: - Providing read-only shard hosting without participating in primary shard allocation or rebalancing - Enabling cheap, ephemeral nodes that can be added/removed without cluster disruption - Supporting integration with object stores (S3, etc.) for pulling index data on-demand ## Why Auto-Expand Replicas 0-all Only The replica-only node type exclusively supports indices with `index.auto_expand_replicas: 0-all` for several critical reasons: 1. **Dynamic Replica Management**: Auto-expand automatically adjusts replica counts when replica-only nodes join/leave, eliminating manual intervention and preventing under-replication 2. **No Manual Rebalancing**: Without auto-expand, adding replica-only nodes would require manual replica count adjustments and could trigger rebalancing on data nodes 3. **Predictable Behavior**: The 0-all setting guarantees one copy per eligible node, making replica distribution deterministic and transparent 4. **Operational Safety**: Prevents accidental allocation of critical production indices to nodes that may be ephemeral or have different SLAs ## High-Level Design ### Core Components 1. **New Node Role (DiscoveryNodeRole.REPLICA_ONLY_ROLE)** - Role name: `replica_only` - Dedicated role that cannot coexist with any other role 2. **Allocation Decider (ReplicaOnlyAllocationDecider)** - Blocks ALL primary shard allocation to replica-only nodes - Blocks replica allocation unless index has auto_expand_replicas: 0-all - Prevents force allocation of primaries (safety guarantee) 3. **Rebalancing Prevention (LocalShardsBalancer)** - Excludes replica-only nodes from rebalancing model entirely - Adding/removing replica-only nodes causes zero data node rebalancing - Maintains cluster balance stability 4. **Replica Promotion Prevention (RoutingNodes)** - Blocks promotion of replicas to primaries on replica-only nodes - Cluster enters YELLOW/RED state when primary fails and only replica-only nodes have copies - Ensures data integrity by requiring regular data node for primaries ## Design Concerns Addressed ### 1. Data Integrity and Cluster Health What happens if primary fails and only replica-only nodes have copies? Replicas on replica-only nodes NEVER promote to primaries. The cluster enters YELLOW/RED state and waits for a regular data node. This prevents data loss scenarios where an ephemeral node becomes the source of truth. ### 2. Rebalancing Isolation Will replica-only nodes trigger rebalancing on production data nodes? Replica-only nodes are completely excluded from the BalancedShardsAllocator model. They are invisible to the balancer, ensuring zero rebalancing impact when nodes join/leave. ### 3. Role Transitions What happens if a data node transitions to replica-only role? - Primary shards are relocated to other data nodes - Replicas from non-auto-expand indices are relocated out - Replicas from auto-expand 0-all indices remain - All transitions are safe with no data loss (canRemain() enforcement) ### 4. Recovery Code Paths Could recovery logic accidentally create primaries on replica-only nodes? No, via multiple layers of protection: - AllocationDecider blocks at allocation time - canForceAllocatePrimary() blocks forced allocation - promoteReplicaToPrimary() has explicit replica-only check ### 5. Auto-Expand Node Counting How do replica-only nodes integrate with auto-expand replica counting? shouldAutoExpandToNode() in ReplicaOnlyAllocationDecider is automatically called by AutoExpandReplicas.getDesiredNumberOfReplicas(). Replica-only nodes are counted only for 0-all indices, ensuring correct replica counts. ## Configuration Example Node configuration: ```yaml node.roles: [replica_only] ``` Index configuration: ```json PUT /my-index { "settings": { "index.auto_expand_replicas": "0-all" } } ```
1 parent 88ffbc7 commit df9b566

8 files changed

Lines changed: 600 additions & 1 deletion

File tree

‎server/src/main/java/org/opensearch/cluster/ClusterModule.java‎

Lines changed: 2 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -77,6 +77,7 @@
7777
import org.opensearch.cluster.routing.allocation.decider.ResizeAllocationDecider;
7878
import org.opensearch.cluster.routing.allocation.decider.RestoreInProgressAllocationDecider;
7979
import org.opensearch.cluster.routing.allocation.decider.SameShardAllocationDecider;
80+
import org.opensearch.cluster.routing.allocation.decider.ReplicaOnlyAllocationDecider;
8081
import org.opensearch.cluster.routing.allocation.decider.SearchReplicaAllocationDecider;
8182
import org.opensearch.cluster.routing.allocation.decider.ShardsLimitAllocationDecider;
8283
import org.opensearch.cluster.routing.allocation.decider.SnapshotInProgressAllocationDecider;
@@ -396,6 +397,7 @@ public static Collection<AllocationDecider> createAllocationDeciders(
396397
addAllocationDecider(deciders, new RestoreInProgressAllocationDecider());
397398
addAllocationDecider(deciders, new FilterAllocationDecider(settings, clusterSettings));
398399
addAllocationDecider(deciders, new SearchReplicaAllocationDecider());
400+
addAllocationDecider(deciders, new ReplicaOnlyAllocationDecider());
399401
addAllocationDecider(deciders, new SameShardAllocationDecider(settings, clusterSettings));
400402
addAllocationDecider(deciders, new DiskThresholdDecider(settings, clusterSettings));
401403
addAllocationDecider(deciders, new WarmDiskThresholdDecider(settings, clusterSettings));

‎server/src/main/java/org/opensearch/cluster/node/DiscoveryNode.java‎

Lines changed: 13 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -130,6 +130,10 @@ public static boolean isDedicatedWarmNode(Settings settings) {
130130
return getRolesFromSettings(settings).stream().allMatch(DiscoveryNodeRole.WARM_ROLE::equals);
131131
}
132132

133+
public static boolean isReplicaOnlyNode(Settings settings) {
134+
return hasRole(settings, DiscoveryNodeRole.REPLICA_ONLY_ROLE);
135+
}
136+
133137
private final String nodeName;
134138
private final String nodeId;
135139
private final String ephemeralId;
@@ -542,6 +546,15 @@ public boolean isSearchNode() {
542546
return roles.contains(DiscoveryNodeRole.SEARCH_ROLE);
543547
}
544548

549+
/**
550+
* Returns whether the node is a replica-only node.
551+
*
552+
* @return true if the node contains a replica_only role, false otherwise
553+
*/
554+
public boolean isReplicaOnlyNode() {
555+
return roles.contains(DiscoveryNodeRole.REPLICA_ONLY_ROLE);
556+
}
557+
545558
/**
546559
* Returns whether the node is a remote store node.
547560
*

‎server/src/main/java/org/opensearch/cluster/node/DiscoveryNodeRole.java‎

Lines changed: 38 additions & 1 deletion
Original file line numberDiff line numberDiff line change
@@ -337,11 +337,48 @@ public void validateRole(List<DiscoveryNodeRole> roles) {
337337

338338
};
339339

340+
/**
341+
* Represents the role for a replica-only node, which hosts replica shards from auto-expand indices
342+
* without participating in primary shard hosting. Replica-only nodes:
343+
* - Never host primary shards or promote replicas to primaries
344+
* - Only accept replica shards from indices with {@code index.auto_expand_replicas: 0-all}
345+
* - Do not trigger rebalancing on regular data nodes when joining/leaving the cluster
346+
* - Must be a dedicated role (cannot coexist with any other role)
347+
*/
348+
public static final DiscoveryNodeRole REPLICA_ONLY_ROLE = new DiscoveryNodeRole("replica_only", "ro", true) {
349+
350+
@Override
351+
public Setting<Boolean> legacySetting() {
352+
// replica_only role is added in 3.5 so doesn't need to configure legacy setting
353+
return null;
354+
}
355+
356+
@Override
357+
public void validateRole(List<DiscoveryNodeRole> roles) {
358+
// replica_only role must be the only role on a node (dedicated)
359+
for (DiscoveryNodeRole role : roles) {
360+
if (role.equals(DiscoveryNodeRole.REPLICA_ONLY_ROLE) == false) {
361+
throw new IllegalArgumentException(
362+
String.format(
363+
Locale.ROOT,
364+
"%s role must be the only role on a node. Cannot be combined with: %s",
365+
DiscoveryNodeRole.REPLICA_ONLY_ROLE.roleName(),
366+
role.roleName()
367+
)
368+
);
369+
}
370+
}
371+
}
372+
373+
};
374+
340375
/**
341376
* The built-in node roles.
342377
*/
343378
public static SortedSet<DiscoveryNodeRole> BUILT_IN_ROLES = Collections.unmodifiableSortedSet(
344-
new TreeSet<>(Arrays.asList(DATA_ROLE, INGEST_ROLE, CLUSTER_MANAGER_ROLE, REMOTE_CLUSTER_CLIENT_ROLE, WARM_ROLE))
379+
new TreeSet<>(
380+
Arrays.asList(DATA_ROLE, INGEST_ROLE, CLUSTER_MANAGER_ROLE, REMOTE_CLUSTER_CLIENT_ROLE, WARM_ROLE, REPLICA_ONLY_ROLE)
381+
)
345382
);
346383

347384
/**

‎server/src/main/java/org/opensearch/cluster/routing/RoutingNodes.java‎

Lines changed: 18 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -32,6 +32,7 @@
3232

3333
package org.opensearch.cluster.routing;
3434

35+
import org.apache.logging.log4j.LogManager;
3536
import org.apache.logging.log4j.Logger;
3637
import org.apache.lucene.util.CollectionUtil;
3738
import org.opensearch.cluster.ClusterState;
@@ -85,6 +86,8 @@
8586
*/
8687
@PublicApi(since = "1.0.0")
8788
public class RoutingNodes implements Iterable<RoutingNode> {
89+
private static final Logger logger = LogManager.getLogger(RoutingNodes.class);
90+
8891
private final Metadata metadata;
8992

9093
private final Map<String, RoutingNode> nodesToShards = new HashMap<>();
@@ -812,6 +815,21 @@ private void promoteReplicaToPrimary(ShardRouting activeReplica, RoutingChangesO
812815
// if the activeReplica was relocating before this call to failShard, its relocation was cancelled earlier when we
813816
// failed initializing replica shards (and moved replica relocation source back to started)
814817
assert activeReplica.started() : "replica relocation should have been cancelled: " + activeReplica;
818+
819+
// CRITICAL: Never promote replicas on replica-only nodes
820+
RoutingNode routingNode = node(activeReplica.currentNodeId());
821+
if (routingNode != null && routingNode.node().isReplicaOnlyNode()) {
822+
logger.warn(
823+
"Cannot promote replica shard [{}] to primary on replica-only node [{}]. "
824+
+ "Shard will remain as replica. Primary must be allocated to a regular data node.",
825+
activeReplica.shardId(),
826+
routingNode.nodeId()
827+
);
828+
// Do NOT call promoteActiveReplicaShardToPrimary - just return
829+
// The primary will remain unassigned, triggering allocation to a data node
830+
return;
831+
}
832+
815833
promoteActiveReplicaShardToPrimary(activeReplica);
816834
routingChangesObserver.replicaPromoted(activeReplica);
817835
}

‎server/src/main/java/org/opensearch/cluster/routing/allocation/allocator/LocalShardsBalancer.java‎

Lines changed: 9 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -765,6 +765,15 @@ MoveDecision decideMove(final ShardRouting shardRouting) {
765765
private Map<String, BalancedShardsAllocator.ModelNode> buildModelFromAssigned() {
766766
Map<String, BalancedShardsAllocator.ModelNode> nodes = new HashMap<>();
767767
for (RoutingNode rn : routingNodes) {
768+
// EXCLUDE replica-only nodes from rebalancing calculations
769+
// These nodes are managed solely by auto-expand replica logic
770+
if (rn.node().isReplicaOnlyNode()) {
771+
if (logger.isTraceEnabled()) {
772+
logger.trace("Excluding replica-only node [{}] from rebalancing model", rn.nodeId());
773+
}
774+
continue;
775+
}
776+
768777
BalancedShardsAllocator.ModelNode node = new BalancedShardsAllocator.ModelNode(rn);
769778
nodes.put(rn.nodeId(), node);
770779
for (ShardRouting shard : rn) {

‎server/src/main/java/org/opensearch/cluster/routing/allocation/decider/ReplicaOnlyAllocationDecider.java‎

Lines changed: 140 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -0,0 +1,140 @@
1+
/*
2+
* SPDX-License-Identifier: Apache-2.0
3+
*
4+
* The OpenSearch Contributors require contributions made to
5+
* this file be licensed under the Apache-2.0 license or a
6+
* compatible open source license.
7+
*/
8+
9+
package org.opensearch.cluster.routing.allocation.decider;
10+
11+
import org.opensearch.cluster.metadata.AutoExpandReplicas;
12+
import org.opensearch.cluster.metadata.IndexMetadata;
13+
import org.opensearch.cluster.node.DiscoveryNode;
14+
import org.opensearch.cluster.routing.RoutingNode;
15+
import org.opensearch.cluster.routing.ShardRouting;
16+
import org.opensearch.cluster.routing.allocation.RoutingAllocation;
17+
18+
/**
19+
* This allocation decider ensures that replica-only nodes follow strict allocation rules:
20+
* <ul>
21+
* <li>Replica-only nodes never host primary shards</li>
22+
* <li>Replica-only nodes only host replica shards from indices with auto_expand_replicas: 0-all</li>
23+
* <li>Primary shards are never promoted on replica-only nodes</li>
24+
* <li>Regular data nodes are not affected by replica-only node presence</li>
25+
* </ul>
26+
*
27+
* @opensearch.internal
28+
*/
29+
public class ReplicaOnlyAllocationDecider extends AllocationDecider {
30+
31+
public static final String NAME = "replica_only";
32+
33+
@Override
34+
public Decision canAllocate(ShardRouting shardRouting, RoutingNode node, RoutingAllocation allocation) {
35+
return canAllocate(shardRouting, node.node(), allocation);
36+
}
37+
38+
@Override
39+
public Decision canRemain(ShardRouting shardRouting, RoutingNode node, RoutingAllocation allocation) {
40+
return canAllocate(shardRouting, node.node(), allocation);
41+
}
42+
43+
@Override
44+
public Decision canForceAllocatePrimary(ShardRouting shardRouting, RoutingNode node, RoutingAllocation allocation) {
45+
// CRITICAL: Never allow primary allocation to replica-only nodes, even with force
46+
if (node.node().isReplicaOnlyNode()) {
47+
return allocation.decision(
48+
Decision.NO,
49+
NAME,
50+
"primary shard [%s] cannot be force allocated to replica-only node [%s]",
51+
shardRouting.shardId(),
52+
node.nodeId()
53+
);
54+
}
55+
return allocation.decision(Decision.YES, NAME, "node is not a replica-only node");
56+
}
57+
58+
@Override
59+
public Decision shouldAutoExpandToNode(IndexMetadata indexMetadata, DiscoveryNode node, RoutingAllocation allocation) {
60+
if (!node.isReplicaOnlyNode()) {
61+
// Regular data nodes participate in auto-expand for all indices
62+
return allocation.decision(Decision.YES, NAME, "node [%s] is a data node, eligible for auto-expand", node.getId());
63+
}
64+
65+
// Replica-only nodes only participate in 0-all auto-expand
66+
AutoExpandReplicas autoExpandReplicas = AutoExpandReplicas.SETTING.get(indexMetadata.getSettings());
67+
boolean isAutoExpandAll = autoExpandReplicas.isEnabled()
68+
&& autoExpandReplicas.getMaxReplicas() == Integer.MAX_VALUE
69+
&& autoExpandReplicas.toString().startsWith("0-");
70+
71+
if (isAutoExpandAll) {
72+
return allocation.decision(
73+
Decision.YES,
74+
NAME,
75+
"replica-only node [%s] is eligible for auto-expand replicas from index [%s] with auto_expand_replicas: 0-all",
76+
node.getId(),
77+
indexMetadata.getIndex().getName()
78+
);
79+
} else {
80+
return allocation.decision(
81+
Decision.NO,
82+
NAME,
83+
"replica-only node [%s] is not eligible for index [%s] without auto_expand_replicas: 0-all",
84+
node.getId(),
85+
indexMetadata.getIndex().getName()
86+
);
87+
}
88+
}
89+
90+
private Decision canAllocate(ShardRouting shardRouting, DiscoveryNode node, RoutingAllocation allocation) {
91+
boolean isReplicaOnlyNode = node.isReplicaOnlyNode();
92+
93+
// Case 1: Primary shard allocation
94+
if (shardRouting.primary()) {
95+
if (isReplicaOnlyNode) {
96+
return allocation.decision(
97+
Decision.NO,
98+
NAME,
99+
"primary shard [%s] cannot be allocated to replica-only node [%s]",
100+
shardRouting.shardId(),
101+
node.getId()
102+
);
103+
}
104+
// Allow primaries on regular data nodes
105+
return allocation.decision(Decision.YES, NAME, "node [%s] is a data node, can host primary shard", node.getId());
106+
}
107+
108+
// Case 2: Replica shard allocation
109+
IndexMetadata indexMetadata = allocation.metadata().getIndexSafe(shardRouting.index());
110+
AutoExpandReplicas autoExpandReplicas = AutoExpandReplicas.SETTING.get(indexMetadata.getSettings());
111+
boolean isAutoExpandAll = autoExpandReplicas.isEnabled()
112+
&& autoExpandReplicas.getMaxReplicas() == Integer.MAX_VALUE
113+
&& autoExpandReplicas.toString().startsWith("0-");
114+
115+
if (isReplicaOnlyNode) {
116+
if (isAutoExpandAll) {
117+
return allocation.decision(
118+
Decision.YES,
119+
NAME,
120+
"replica shard [%s] from auto-expand (0-all) index can be allocated to replica-only node [%s]",
121+
shardRouting.shardId(),
122+
node.getId()
123+
);
124+
} else {
125+
return allocation.decision(
126+
Decision.NO,
127+
NAME,
128+
"replica shard [%s] cannot be allocated to replica-only node [%s] "
129+
+ "because index [%s] does not have auto_expand_replicas: 0-all",
130+
shardRouting.shardId(),
131+
node.getId(),
132+
indexMetadata.getIndex().getName()
133+
);
134+
}
135+
}
136+
137+
// Regular data nodes can host any replica
138+
return allocation.decision(Decision.YES, NAME, "node [%s] is a data node, can host replica shard", node.getId());
139+
}
140+
}

‎server/src/test/java/org/opensearch/cluster/node/DiscoveryNodeRoleTests.java‎

Lines changed: 37 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -159,4 +159,41 @@ public void testRoleNameIsCaseInsensitive() {
159159
assertEquals(roleName.toLowerCase(Locale.ROOT), dynamicRole.roleName());
160160
assertEquals(roleNameAbbreviation.toLowerCase(Locale.ROOT), dynamicRole.roleNameAbbreviation());
161161
}
162+
163+
public void testReplicaOnlyRoleIsDedicated() {
164+
// replica_only role cannot coexist with any other role
165+
final IllegalArgumentException e1 = expectThrows(
166+
IllegalArgumentException.class,
167+
() -> DiscoveryNodeRole.REPLICA_ONLY_ROLE.validateRole(Arrays.asList(DiscoveryNodeRole.REPLICA_ONLY_ROLE, DiscoveryNodeRole.DATA_ROLE))
168+
);
169+
assertThat(e1, hasToString(containsString("replica_only role must be the only role")));
170+
171+
final IllegalArgumentException e2 = expectThrows(
172+
IllegalArgumentException.class,
173+
() -> DiscoveryNodeRole.REPLICA_ONLY_ROLE.validateRole(
174+
Arrays.asList(DiscoveryNodeRole.REPLICA_ONLY_ROLE, DiscoveryNodeRole.CLUSTER_MANAGER_ROLE)
175+
)
176+
);
177+
assertThat(e2, hasToString(containsString("replica_only role must be the only role")));
178+
179+
final IllegalArgumentException e3 = expectThrows(
180+
IllegalArgumentException.class,
181+
() -> DiscoveryNodeRole.REPLICA_ONLY_ROLE.validateRole(Arrays.asList(DiscoveryNodeRole.REPLICA_ONLY_ROLE, DiscoveryNodeRole.INGEST_ROLE))
182+
);
183+
assertThat(e3, hasToString(containsString("replica_only role must be the only role")));
184+
185+
// replica_only role by itself should not throw
186+
DiscoveryNodeRole.REPLICA_ONLY_ROLE.validateRole(Arrays.asList(DiscoveryNodeRole.REPLICA_ONLY_ROLE));
187+
}
188+
189+
public void testReplicaOnlyRoleProperties() {
190+
assertEquals("replica_only", DiscoveryNodeRole.REPLICA_ONLY_ROLE.roleName());
191+
assertEquals("ro", DiscoveryNodeRole.REPLICA_ONLY_ROLE.roleNameAbbreviation());
192+
assertTrue(DiscoveryNodeRole.REPLICA_ONLY_ROLE.canContainData());
193+
assertNull(DiscoveryNodeRole.REPLICA_ONLY_ROLE.legacySetting());
194+
}
195+
196+
public void testReplicaOnlyRoleInBuiltInRoles() {
197+
assertTrue(DiscoveryNodeRole.BUILT_IN_ROLES.contains(DiscoveryNodeRole.REPLICA_ONLY_ROLE));
198+
}
162199
}

0 commit comments

Comments
 (0)