Skip to content

Commit ebadf42

Browse files
author
Swetha Guptha
committed
Optimise memory for rest cluster health calls with inline shard aggregations for levels cluster and indices.
Signed-off-by: Swetha Guptha <[email protected]>
1 parent ed65482 commit ebadf42

File tree

11 files changed

+433
-24
lines changed

11 files changed

+433
-24
lines changed

‎server/src/internalClusterTest/java/org/opensearch/cluster/ClusterHealthIT.java

+129
Original file line numberDiff line numberDiff line change
@@ -37,6 +37,8 @@
3737
import org.opensearch.action.support.IndicesOptions;
3838
import org.opensearch.action.support.PlainActionFuture;
3939
import org.opensearch.cluster.health.ClusterHealthStatus;
40+
import org.opensearch.cluster.health.ClusterIndexHealth;
41+
import org.opensearch.cluster.health.ClusterShardHealth;
4042
import org.opensearch.cluster.metadata.IndexMetadata;
4143
import org.opensearch.cluster.routing.UnassignedInfo;
4244
import org.opensearch.cluster.service.ClusterService;
@@ -49,6 +51,7 @@
4951

5052
import java.util.ArrayList;
5153
import java.util.List;
54+
import java.util.Map;
5255
import java.util.concurrent.atomic.AtomicBoolean;
5356

5457
import static org.opensearch.test.hamcrest.OpenSearchAssertions.assertAcked;
@@ -439,4 +442,130 @@ public void clusterStateProcessed(String source, ClusterState oldState, ClusterS
439442
completionFuture.actionGet(TimeValue.timeValueSeconds(30));
440443
}
441444
}
445+
446+
public void testHealthAtLevelParamEqualsClusterIsHonoured() {
447+
createIndex(
448+
"test1",
449+
Settings.builder().put(IndexMetadata.SETTING_NUMBER_OF_SHARDS, 1).put(IndexMetadata.SETTING_NUMBER_OF_REPLICAS, 0).build()
450+
);
451+
ensureGreen();
452+
ClusterHealthResponse healthResponse = client().admin()
453+
.cluster()
454+
.prepareHealth()
455+
.setHonourClusterHealthLevel(true)
456+
.execute()
457+
.actionGet();
458+
assertEquals(healthResponse.getStatus(), ClusterHealthStatus.GREEN);
459+
assertTrue(healthResponse.getIndices().isEmpty());
460+
assertEquals(1, healthResponse.getActiveShards());
461+
assertEquals(1, healthResponse.getActivePrimaryShards());
462+
assertEquals(0, healthResponse.getUnassignedShards());
463+
assertEquals(0, healthResponse.getInitializingShards());
464+
assertEquals(0, healthResponse.getRelocatingShards());
465+
assertEquals(0, healthResponse.getDelayedUnassignedShards());
466+
}
467+
468+
public void testHealthAtLevelParamEqualIndicesIsHonoured() {
469+
createIndex(
470+
"test1",
471+
Settings.builder().put(IndexMetadata.SETTING_NUMBER_OF_SHARDS, 1).put(IndexMetadata.SETTING_NUMBER_OF_REPLICAS, 0).build()
472+
);
473+
ensureGreen();
474+
ClusterHealthResponse healthResponse = client().admin()
475+
.cluster()
476+
.prepareHealth()
477+
.setLevel("indices")
478+
.setHonourClusterHealthLevel(true)
479+
.execute()
480+
.actionGet();
481+
assertEquals(healthResponse.getStatus(), ClusterHealthStatus.GREEN);
482+
483+
assertEquals(1, healthResponse.getActiveShards());
484+
assertEquals(1, healthResponse.getActivePrimaryShards());
485+
assertEquals(0, healthResponse.getUnassignedShards());
486+
assertEquals(0, healthResponse.getInitializingShards());
487+
assertEquals(0, healthResponse.getRelocatingShards());
488+
assertEquals(0, healthResponse.getDelayedUnassignedShards());
489+
490+
Map<String, ClusterIndexHealth> indices = healthResponse.getIndices();
491+
assertFalse(indices.isEmpty());
492+
assertEquals(1, indices.size());
493+
for (Map.Entry<String, ClusterIndexHealth> indicesHealth : indices.entrySet()) {
494+
String indexName = indicesHealth.getKey();
495+
assertEquals("test1", indexName);
496+
ClusterIndexHealth indicesHealthValue = indicesHealth.getValue();
497+
assertEquals(1, indicesHealthValue.getActiveShards());
498+
assertEquals(1, indicesHealthValue.getActivePrimaryShards());
499+
assertEquals(0, indicesHealthValue.getInitializingShards());
500+
assertEquals(0, indicesHealthValue.getUnassignedShards());
501+
assertEquals(0, indicesHealthValue.getDelayedUnassignedShards());
502+
assertEquals(0, indicesHealthValue.getRelocatingShards());
503+
assertEquals(ClusterHealthStatus.GREEN, indicesHealthValue.getStatus());
504+
assertTrue(indicesHealthValue.getShards().isEmpty());
505+
}
506+
}
507+
508+
public void testHealthAtLevelParamEqualShardsIsHonoured() {
509+
createIndex(
510+
"test1",
511+
Settings.builder().put(IndexMetadata.SETTING_NUMBER_OF_SHARDS, 1).put(IndexMetadata.SETTING_NUMBER_OF_REPLICAS, 1).build()
512+
);
513+
ensureGreen(TimeValue.timeValueSeconds(120), "test1");
514+
createIndex(
515+
"test2",
516+
Settings.builder().put(IndexMetadata.SETTING_NUMBER_OF_SHARDS, 1).put(IndexMetadata.SETTING_NUMBER_OF_REPLICAS, 1).build()
517+
);
518+
ensureGreen(TimeValue.timeValueSeconds(120));
519+
client().admin()
520+
.indices()
521+
.prepareUpdateSettings()
522+
.setIndices("test2")
523+
.setSettings(Settings.builder().put("index.number_of_replicas", 2).build())
524+
.execute()
525+
.actionGet();
526+
ClusterHealthResponse healthResponse = client().admin()
527+
.cluster()
528+
.prepareHealth()
529+
.setLevel("shards")
530+
.setHonourClusterHealthLevel(true)
531+
.execute()
532+
.actionGet();
533+
assertEquals(healthResponse.getStatus(), ClusterHealthStatus.YELLOW);
534+
535+
assertEquals(4, healthResponse.getActiveShards());
536+
assertEquals(2, healthResponse.getActivePrimaryShards());
537+
assertEquals(1, healthResponse.getUnassignedShards());
538+
assertEquals(0, healthResponse.getInitializingShards());
539+
assertEquals(0, healthResponse.getRelocatingShards());
540+
assertEquals(0, healthResponse.getDelayedUnassignedShards());
541+
542+
Map<String, ClusterIndexHealth> indices = healthResponse.getIndices();
543+
assertFalse(indices.isEmpty());
544+
assertEquals(2, indices.size());
545+
for (Map.Entry<String, ClusterIndexHealth> indicesHealth : indices.entrySet()) {
546+
String indexName = indicesHealth.getKey();
547+
boolean indexHasMoreReplicas = indexName.equals("test2");
548+
ClusterIndexHealth indicesHealthValue = indicesHealth.getValue();
549+
assertEquals(2, indicesHealthValue.getActiveShards());
550+
assertEquals(1, indicesHealthValue.getActivePrimaryShards());
551+
assertEquals(0, indicesHealthValue.getInitializingShards());
552+
assertEquals(indexHasMoreReplicas ? 1 : 0, indicesHealthValue.getUnassignedShards());
553+
assertEquals(0, indicesHealthValue.getDelayedUnassignedShards());
554+
assertEquals(0, indicesHealthValue.getRelocatingShards());
555+
assertEquals(indexHasMoreReplicas ? ClusterHealthStatus.YELLOW : ClusterHealthStatus.GREEN, indicesHealthValue.getStatus());
556+
Map<Integer, ClusterShardHealth> shards = indicesHealthValue.getShards();
557+
assertFalse(shards.isEmpty());
558+
assertEquals(1, shards.size());
559+
for (Map.Entry<Integer, ClusterShardHealth> shardHealth : shards.entrySet()) {
560+
ClusterShardHealth clusterShardHealth = shardHealth.getValue();
561+
assertEquals(2, clusterShardHealth.getActiveShards());
562+
assertEquals(indexHasMoreReplicas ? 1 : 0, clusterShardHealth.getUnassignedShards());
563+
assertEquals(0, clusterShardHealth.getDelayedUnassignedShards());
564+
assertEquals(0, clusterShardHealth.getRelocatingShards());
565+
assertEquals(0, clusterShardHealth.getInitializingShards());
566+
assertTrue(clusterShardHealth.isPrimaryActive());
567+
assertEquals(indexHasMoreReplicas ? ClusterHealthStatus.YELLOW : ClusterHealthStatus.GREEN, clusterShardHealth.getStatus());
568+
}
569+
}
570+
}
442571
}

‎server/src/main/java/org/opensearch/action/admin/cluster/health/ClusterHealthRequest.java

+25
Original file line numberDiff line numberDiff line change
@@ -76,6 +76,17 @@ public class ClusterHealthRequest extends ClusterManagerNodeReadRequest<ClusterH
7676
*/
7777
private Level level = Level.CLUSTER;
7878

79+
/**
80+
* This flag will be used by the TransportClusterHealthAction whether to return indices/shards info in the ClusterHealthResponse or not.
81+
* When the flag is disabled - indices/shard info will be returned in ClusterHealthResponse regardless of the health level received in the request.
82+
* When the flag is enabled - indices/shards info will be set according to health level received in the request.
83+
* For Level.CLUSTER, information on indices/shards will NOT be returned to the transport client
84+
* For Level.INDICES, information on indices will be returned to the transport client.
85+
* For Level.SHARDS, information on indices and shards will be returned to the transport client
86+
* By default, the flag is disabled.
87+
*/
88+
private boolean honourClusterHealthLevel = false;
89+
7990
public ClusterHealthRequest() {}
8091

8192
public ClusterHealthRequest(String... indices) {
@@ -104,6 +115,9 @@ public ClusterHealthRequest(StreamInput in) throws IOException {
104115
if (in.getVersion().onOrAfter(Version.V_2_6_0)) {
105116
ensureNodeWeighedIn = in.readBoolean();
106117
}
118+
if (in.getVersion().onOrAfter(Version.V_3_0_0)) {
119+
honourClusterHealthLevel = in.readBoolean();
120+
}
107121
}
108122

109123
@Override
@@ -139,6 +153,9 @@ public void writeTo(StreamOutput out) throws IOException {
139153
if (out.getVersion().onOrAfter(Version.V_2_6_0)) {
140154
out.writeBoolean(ensureNodeWeighedIn);
141155
}
156+
if (out.getVersion().onOrAfter(Version.V_3_0_0)) {
157+
out.writeBoolean(honourClusterHealthLevel);
158+
}
142159
}
143160

144161
@Override
@@ -337,6 +354,14 @@ public final boolean ensureNodeWeighedIn() {
337354
return ensureNodeWeighedIn;
338355
}
339356

357+
public boolean isHonourClusterHealthLevel() {
358+
return honourClusterHealthLevel;
359+
}
360+
361+
public void setHonourClusterHealthLevel(boolean honourClusterHealthLevel) {
362+
this.honourClusterHealthLevel = honourClusterHealthLevel;
363+
}
364+
340365
@Override
341366
public ActionRequestValidationException validate() {
342367
if (level.equals(Level.AWARENESS_ATTRIBUTES) && indices.length > 0) {

‎server/src/main/java/org/opensearch/action/admin/cluster/health/ClusterHealthRequestBuilder.java

+5
Original file line numberDiff line numberDiff line change
@@ -171,4 +171,9 @@ public final ClusterHealthRequestBuilder setEnsureNodeWeighedIn(boolean ensureNo
171171
request.ensureNodeWeighedIn(ensureNodeCommissioned);
172172
return this;
173173
}
174+
175+
public ClusterHealthRequestBuilder setHonourClusterHealthLevel(boolean honourClusterHealthLevel) {
176+
request.setHonourClusterHealthLevel(honourClusterHealthLevel);
177+
return this;
178+
}
174179
}

‎server/src/main/java/org/opensearch/action/admin/cluster/health/ClusterHealthResponse.java

+45
Original file line numberDiff line numberDiff line change
@@ -237,6 +237,28 @@ public ClusterHealthResponse(
237237
this.clusterHealthStatus = clusterStateHealth.getStatus();
238238
}
239239

240+
public ClusterHealthResponse(
241+
String clusterName,
242+
String[] concreteIndices,
243+
ClusterHealthRequest clusterHealthRequest,
244+
ClusterState clusterState,
245+
int numberOfPendingTasks,
246+
int numberOfInFlightFetch,
247+
TimeValue taskMaxWaitingTime
248+
) {
249+
this.clusterName = clusterName;
250+
this.numberOfPendingTasks = numberOfPendingTasks;
251+
this.numberOfInFlightFetch = numberOfInFlightFetch;
252+
this.taskMaxWaitingTime = taskMaxWaitingTime;
253+
if (clusterHealthRequest.isHonourClusterHealthLevel()) {
254+
this.clusterStateHealth = new ClusterStateHealth(clusterState, concreteIndices, clusterHealthRequest.level());
255+
} else {
256+
this.clusterStateHealth = new ClusterStateHealth(clusterState, concreteIndices);
257+
}
258+
this.clusterHealthStatus = clusterStateHealth.getStatus();
259+
this.delayedUnassignedShards = clusterStateHealth.getDelayedUnassignedShards();
260+
}
261+
240262
// Awareness Attribute health
241263
public ClusterHealthResponse(
242264
String clusterName,
@@ -261,6 +283,29 @@ public ClusterHealthResponse(
261283
this.clusterAwarenessHealth = new ClusterAwarenessHealth(clusterState, clusterSettings, awarenessAttributeName);
262284
}
263285

286+
public ClusterHealthResponse(
287+
String clusterName,
288+
ClusterHealthRequest clusterHealthRequest,
289+
ClusterState clusterState,
290+
ClusterSettings clusterSettings,
291+
String[] concreteIndices,
292+
String awarenessAttributeName,
293+
int numberOfPendingTasks,
294+
int numberOfInFlightFetch,
295+
TimeValue taskMaxWaitingTime
296+
) {
297+
this(
298+
clusterName,
299+
concreteIndices,
300+
clusterHealthRequest,
301+
clusterState,
302+
numberOfPendingTasks,
303+
numberOfInFlightFetch,
304+
taskMaxWaitingTime
305+
);
306+
this.clusterAwarenessHealth = new ClusterAwarenessHealth(clusterState, clusterSettings, awarenessAttributeName);
307+
}
308+
264309
/**
265310
* For XContent Parser and serialization tests
266311
*/

‎server/src/main/java/org/opensearch/action/admin/cluster/health/TransportClusterHealthAction.java

+3-4
Original file line numberDiff line numberDiff line change
@@ -52,7 +52,6 @@
5252
import org.opensearch.cluster.metadata.ProcessClusterEventTimeoutException;
5353
import org.opensearch.cluster.node.DiscoveryNode;
5454
import org.opensearch.cluster.routing.NodeWeighedAwayException;
55-
import org.opensearch.cluster.routing.UnassignedInfo;
5655
import org.opensearch.cluster.routing.WeightedRoutingUtils;
5756
import org.opensearch.cluster.routing.allocation.AllocationService;
5857
import org.opensearch.cluster.service.ClusterService;
@@ -496,13 +495,13 @@ private ClusterHealthResponse clusterHealth(
496495
concreteIndices = clusterState.getMetadata().getConcreteAllIndices();
497496
return new ClusterHealthResponse(
498497
clusterState.getClusterName().value(),
498+
request,
499499
clusterState,
500500
clusterService.getClusterSettings(),
501501
concreteIndices,
502502
awarenessAttribute,
503503
numberOfPendingTasks,
504504
numberOfInFlightFetch,
505-
UnassignedInfo.getNumberOfDelayedUnassigned(clusterState),
506505
pendingTaskTimeInQueue
507506
);
508507
}
@@ -514,10 +513,10 @@ private ClusterHealthResponse clusterHealth(
514513
ClusterHealthResponse response = new ClusterHealthResponse(
515514
clusterState.getClusterName().value(),
516515
Strings.EMPTY_ARRAY,
516+
request,
517517
clusterState,
518518
numberOfPendingTasks,
519519
numberOfInFlightFetch,
520-
UnassignedInfo.getNumberOfDelayedUnassigned(clusterState),
521520
pendingTaskTimeInQueue
522521
);
523522
response.setStatus(ClusterHealthStatus.RED);
@@ -527,10 +526,10 @@ private ClusterHealthResponse clusterHealth(
527526
return new ClusterHealthResponse(
528527
clusterState.getClusterName().value(),
529528
concreteIndices,
529+
request,
530530
clusterState,
531531
numberOfPendingTasks,
532532
numberOfInFlightFetch,
533-
UnassignedInfo.getNumberOfDelayedUnassigned(clusterState),
534533
pendingTaskTimeInQueue
535534
);
536535
}

‎server/src/main/java/org/opensearch/action/admin/cluster/stats/TransportClusterStatsAction.java

+2-1
Original file line numberDiff line numberDiff line change
@@ -34,6 +34,7 @@
3434

3535
import org.apache.lucene.store.AlreadyClosedException;
3636
import org.opensearch.action.FailedNodeException;
37+
import org.opensearch.action.admin.cluster.health.ClusterHealthRequest;
3738
import org.opensearch.action.admin.cluster.node.info.NodeInfo;
3839
import org.opensearch.action.admin.cluster.node.stats.NodeStats;
3940
import org.opensearch.action.admin.indices.stats.CommonStats;
@@ -209,7 +210,7 @@ protected ClusterStatsNodeResponse nodeOperation(ClusterStatsNodeRequest nodeReq
209210

210211
ClusterHealthStatus clusterStatus = null;
211212
if (clusterService.state().nodes().isLocalNodeElectedClusterManager()) {
212-
clusterStatus = new ClusterStateHealth(clusterService.state()).getStatus();
213+
clusterStatus = new ClusterStateHealth(clusterService.state(), ClusterHealthRequest.Level.CLUSTER).getStatus();
213214
}
214215

215216
return new ClusterStatsNodeResponse(

0 commit comments

Comments
 (0)