diff --git a/CHANGELOG.md b/CHANGELOG.md index a8df628813839..332dad2a7370a 100644 --- a/CHANGELOG.md +++ b/CHANGELOG.md @@ -26,8 +26,9 @@ The format is based on [Keep a Changelog](https://keepachangelog.com/en/1.0.0/), - Add _list/shards API as paginated alternate to _cat/shards ([#14641](https://github.com/opensearch-project/OpenSearch/pull/14641)) - Latency and Memory allocation improvements to Multi Term Aggregation queries ([#14993](https://github.com/opensearch-project/OpenSearch/pull/14993)) - Flat object field use IndexOrDocValuesQuery to optimize query ([#14383](https://github.com/opensearch-project/OpenSearch/issues/14383)) -- Add method to return dynamic SecureTransportParameters from SecureTransportSettingsProvider interface ([#16387](https://github.com/opensearch-project/OpenSearch/pull/16387) - Add support for renaming aliases during snapshot restore ([#16292](https://github.com/opensearch-project/OpenSearch/pull/16292)) +- Add method to return dynamic SecureTransportParameters from SecureTransportSettingsProvider interface ([#16387](https://github.com/opensearch-project/OpenSearch/pull/16387)) +- URI path filtering support in cluster stats API ([#15938](https://github.com/opensearch-project/OpenSearch/pull/15938)) ### Dependencies - Bump `com.azure:azure-identity` from 1.13.0 to 1.13.2 ([#15578](https://github.com/opensearch-project/OpenSearch/pull/15578)) @@ -66,6 +67,7 @@ The format is based on [Keep a Changelog](https://keepachangelog.com/en/1.0.0/), - Code cleanup: Remove ApproximateIndexOrDocValuesQuery ([#16273](https://github.com/opensearch-project/OpenSearch/pull/16273)) - Optimise clone operation for incremental full cluster snapshots ([#16296](https://github.com/opensearch-project/OpenSearch/pull/16296)) - Update last seen cluster state in the commit phase ([#16215](https://github.com/opensearch-project/OpenSearch/pull/16215)) +- Make multiple settings dynamic for tuning on larger clusters([#16347](https://github.com/opensearch-project/OpenSearch/pull/16347)) ### Deprecated @@ -86,11 +88,13 @@ The format is based on [Keep a Changelog](https://keepachangelog.com/en/1.0.0/), - Fix multi-search with template doesn't return status code ([#16265](https://github.com/opensearch-project/OpenSearch/pull/16265)) - [Streaming Indexing] Fix intermittent 'The bulk request must be terminated by a newline [\n]' failures [#16337](https://github.com/opensearch-project/OpenSearch/pull/16337)) - Fix wrong default value when setting `index.number_of_routing_shards` to null on index creation ([#16331](https://github.com/opensearch-project/OpenSearch/pull/16331)) -- [Workload Management] Make query groups persistent across process restarts [#16370](https://github.com/opensearch-project/OpenSearch/pull/16370) +- [Workload Management] Make query groups persistent across process restarts ([#16370](https://github.com/opensearch-project/OpenSearch/pull/16370)) +- [Workload Management] Enhance rejection mechanism in workload management ([#16417](https://github.com/opensearch-project/OpenSearch/pull/16417)) - Fix inefficient Stream API call chains ending with count() ([#15386](https://github.com/opensearch-project/OpenSearch/pull/15386)) - Fix array hashCode calculation in ResyncReplicationRequest ([#16378](https://github.com/opensearch-project/OpenSearch/pull/16378)) - Fix missing fields in task index mapping to ensure proper task result storage ([#16201](https://github.com/opensearch-project/OpenSearch/pull/16201)) - Fix typo super->sb in method toString() of RemoteStoreNodeAttribute ([#15362](https://github.com/opensearch-project/OpenSearch/pull/15362)) +- [Workload Management] Fixing Create/Update QueryGroup TransportActions to execute from non-cluster manager nodes ([16422](https://github.com/opensearch-project/OpenSearch/pull/16422)) ### Security diff --git a/buildSrc/src/main/java/org/opensearch/gradle/test/DistroTestPlugin.java b/buildSrc/src/main/java/org/opensearch/gradle/test/DistroTestPlugin.java index 9365f1c732229..439d0de39584d 100644 --- a/buildSrc/src/main/java/org/opensearch/gradle/test/DistroTestPlugin.java +++ b/buildSrc/src/main/java/org/opensearch/gradle/test/DistroTestPlugin.java @@ -77,9 +77,9 @@ import java.util.stream.Stream; public class DistroTestPlugin implements Plugin { - private static final String SYSTEM_JDK_VERSION = "23+37"; + private static final String SYSTEM_JDK_VERSION = "23.0.1+11"; private static final String SYSTEM_JDK_VENDOR = "adoptium"; - private static final String GRADLE_JDK_VERSION = "23+37"; + private static final String GRADLE_JDK_VERSION = "23.0.1+11"; private static final String GRADLE_JDK_VENDOR = "adoptium"; // all distributions used by distro tests. this is temporary until tests are per distribution diff --git a/buildSrc/version.properties b/buildSrc/version.properties index 5740c124910b9..f9a8bee5783b1 100644 --- a/buildSrc/version.properties +++ b/buildSrc/version.properties @@ -2,7 +2,7 @@ opensearch = 3.0.0 lucene = 9.12.0 bundled_jdk_vendor = adoptium -bundled_jdk = 23+37 +bundled_jdk = 23.0.1+11 # optional dependencies spatial4j = 0.7 diff --git a/plugins/workload-management/src/main/java/org/opensearch/plugin/wlm/action/CreateQueryGroupRequest.java b/plugins/workload-management/src/main/java/org/opensearch/plugin/wlm/action/CreateQueryGroupRequest.java index d92283391dd3b..1ce04faa7ccc1 100644 --- a/plugins/workload-management/src/main/java/org/opensearch/plugin/wlm/action/CreateQueryGroupRequest.java +++ b/plugins/workload-management/src/main/java/org/opensearch/plugin/wlm/action/CreateQueryGroupRequest.java @@ -8,8 +8,8 @@ package org.opensearch.plugin.wlm.action; -import org.opensearch.action.ActionRequest; import org.opensearch.action.ActionRequestValidationException; +import org.opensearch.action.support.clustermanager.ClusterManagerNodeRequest; import org.opensearch.cluster.metadata.QueryGroup; import org.opensearch.common.UUIDs; import org.opensearch.core.common.io.stream.StreamInput; @@ -33,7 +33,7 @@ * * @opensearch.experimental */ -public class CreateQueryGroupRequest extends ActionRequest { +public class CreateQueryGroupRequest extends ClusterManagerNodeRequest { private final QueryGroup queryGroup; /** diff --git a/plugins/workload-management/src/main/java/org/opensearch/plugin/wlm/action/TransportCreateQueryGroupAction.java b/plugins/workload-management/src/main/java/org/opensearch/plugin/wlm/action/TransportCreateQueryGroupAction.java index 190ff17261bb4..dff9c429d63b0 100644 --- a/plugins/workload-management/src/main/java/org/opensearch/plugin/wlm/action/TransportCreateQueryGroupAction.java +++ b/plugins/workload-management/src/main/java/org/opensearch/plugin/wlm/action/TransportCreateQueryGroupAction.java @@ -9,43 +9,82 @@ package org.opensearch.plugin.wlm.action; import org.opensearch.action.support.ActionFilters; -import org.opensearch.action.support.HandledTransportAction; +import org.opensearch.action.support.clustermanager.TransportClusterManagerNodeAction; +import org.opensearch.cluster.ClusterState; +import org.opensearch.cluster.block.ClusterBlockException; +import org.opensearch.cluster.block.ClusterBlockLevel; +import org.opensearch.cluster.metadata.IndexNameExpressionResolver; import org.opensearch.common.inject.Inject; import org.opensearch.core.action.ActionListener; +import org.opensearch.core.common.io.stream.StreamInput; import org.opensearch.plugin.wlm.service.QueryGroupPersistenceService; -import org.opensearch.tasks.Task; +import org.opensearch.threadpool.ThreadPool; import org.opensearch.transport.TransportService; +import java.io.IOException; + +import static org.opensearch.threadpool.ThreadPool.Names.SAME; + /** * Transport action to create QueryGroup * * @opensearch.experimental */ -public class TransportCreateQueryGroupAction extends HandledTransportAction { +public class TransportCreateQueryGroupAction extends TransportClusterManagerNodeAction { private final QueryGroupPersistenceService queryGroupPersistenceService; /** * Constructor for TransportCreateQueryGroupAction * - * @param actionName - action name + * @param threadPool - {@link ThreadPool} object * @param transportService - a {@link TransportService} object * @param actionFilters - a {@link ActionFilters} object + * @param indexNameExpressionResolver - {@link IndexNameExpressionResolver} object * @param queryGroupPersistenceService - a {@link QueryGroupPersistenceService} object */ @Inject public TransportCreateQueryGroupAction( - String actionName, + ThreadPool threadPool, TransportService transportService, ActionFilters actionFilters, + IndexNameExpressionResolver indexNameExpressionResolver, QueryGroupPersistenceService queryGroupPersistenceService ) { - super(CreateQueryGroupAction.NAME, transportService, actionFilters, CreateQueryGroupRequest::new); + super( + CreateQueryGroupAction.NAME, + transportService, + queryGroupPersistenceService.getClusterService(), + threadPool, + actionFilters, + CreateQueryGroupRequest::new, + indexNameExpressionResolver + ); this.queryGroupPersistenceService = queryGroupPersistenceService; } @Override - protected void doExecute(Task task, CreateQueryGroupRequest request, ActionListener listener) { + protected void clusterManagerOperation( + CreateQueryGroupRequest request, + ClusterState clusterState, + ActionListener listener + ) { queryGroupPersistenceService.persistInClusterStateMetadata(request.getQueryGroup(), listener); } + + @Override + protected String executor() { + return SAME; + } + + @Override + protected CreateQueryGroupResponse read(StreamInput in) throws IOException { + return new CreateQueryGroupResponse(in); + } + + @Override + protected ClusterBlockException checkBlock(CreateQueryGroupRequest request, ClusterState state) { + return state.blocks().globalBlockedException(ClusterBlockLevel.METADATA_WRITE); + } + } diff --git a/plugins/workload-management/src/main/java/org/opensearch/plugin/wlm/action/TransportUpdateQueryGroupAction.java b/plugins/workload-management/src/main/java/org/opensearch/plugin/wlm/action/TransportUpdateQueryGroupAction.java index a6aa2da8fdc08..09a0da7086b36 100644 --- a/plugins/workload-management/src/main/java/org/opensearch/plugin/wlm/action/TransportUpdateQueryGroupAction.java +++ b/plugins/workload-management/src/main/java/org/opensearch/plugin/wlm/action/TransportUpdateQueryGroupAction.java @@ -9,43 +9,81 @@ package org.opensearch.plugin.wlm.action; import org.opensearch.action.support.ActionFilters; -import org.opensearch.action.support.HandledTransportAction; +import org.opensearch.action.support.clustermanager.TransportClusterManagerNodeAction; +import org.opensearch.cluster.ClusterState; +import org.opensearch.cluster.block.ClusterBlockException; +import org.opensearch.cluster.block.ClusterBlockLevel; +import org.opensearch.cluster.metadata.IndexNameExpressionResolver; import org.opensearch.common.inject.Inject; import org.opensearch.core.action.ActionListener; +import org.opensearch.core.common.io.stream.StreamInput; import org.opensearch.plugin.wlm.service.QueryGroupPersistenceService; -import org.opensearch.tasks.Task; +import org.opensearch.threadpool.ThreadPool; import org.opensearch.transport.TransportService; +import java.io.IOException; + +import static org.opensearch.threadpool.ThreadPool.Names.SAME; + /** * Transport action to update QueryGroup * * @opensearch.experimental */ -public class TransportUpdateQueryGroupAction extends HandledTransportAction { +public class TransportUpdateQueryGroupAction extends TransportClusterManagerNodeAction { private final QueryGroupPersistenceService queryGroupPersistenceService; /** * Constructor for TransportUpdateQueryGroupAction * - * @param actionName - action name + * @param threadPool - {@link ThreadPool} object * @param transportService - a {@link TransportService} object * @param actionFilters - a {@link ActionFilters} object + * @param indexNameExpressionResolver - {@link IndexNameExpressionResolver} object * @param queryGroupPersistenceService - a {@link QueryGroupPersistenceService} object */ @Inject public TransportUpdateQueryGroupAction( - String actionName, + ThreadPool threadPool, TransportService transportService, ActionFilters actionFilters, + IndexNameExpressionResolver indexNameExpressionResolver, QueryGroupPersistenceService queryGroupPersistenceService ) { - super(UpdateQueryGroupAction.NAME, transportService, actionFilters, UpdateQueryGroupRequest::new); + super( + UpdateQueryGroupAction.NAME, + transportService, + queryGroupPersistenceService.getClusterService(), + threadPool, + actionFilters, + UpdateQueryGroupRequest::new, + indexNameExpressionResolver + ); this.queryGroupPersistenceService = queryGroupPersistenceService; } @Override - protected void doExecute(Task task, UpdateQueryGroupRequest request, ActionListener listener) { + protected void clusterManagerOperation( + UpdateQueryGroupRequest request, + ClusterState clusterState, + ActionListener listener + ) { queryGroupPersistenceService.updateInClusterStateMetadata(request, listener); } + + @Override + protected String executor() { + return SAME; + } + + @Override + protected UpdateQueryGroupResponse read(StreamInput in) throws IOException { + return new UpdateQueryGroupResponse(in); + } + + @Override + protected ClusterBlockException checkBlock(UpdateQueryGroupRequest request, ClusterState state) { + return state.blocks().globalBlockedException(ClusterBlockLevel.METADATA_WRITE); + } } diff --git a/plugins/workload-management/src/main/java/org/opensearch/plugin/wlm/action/UpdateQueryGroupRequest.java b/plugins/workload-management/src/main/java/org/opensearch/plugin/wlm/action/UpdateQueryGroupRequest.java index 048b599f095fd..18af58289be13 100644 --- a/plugins/workload-management/src/main/java/org/opensearch/plugin/wlm/action/UpdateQueryGroupRequest.java +++ b/plugins/workload-management/src/main/java/org/opensearch/plugin/wlm/action/UpdateQueryGroupRequest.java @@ -8,8 +8,8 @@ package org.opensearch.plugin.wlm.action; -import org.opensearch.action.ActionRequest; import org.opensearch.action.ActionRequestValidationException; +import org.opensearch.action.support.clustermanager.ClusterManagerNodeRequest; import org.opensearch.cluster.metadata.QueryGroup; import org.opensearch.core.common.io.stream.StreamInput; import org.opensearch.core.common.io.stream.StreamOutput; @@ -23,7 +23,7 @@ * * @opensearch.experimental */ -public class UpdateQueryGroupRequest extends ActionRequest { +public class UpdateQueryGroupRequest extends ClusterManagerNodeRequest { private final String name; private final MutableQueryGroupFragment mutableQueryGroupFragment; diff --git a/qa/rolling-upgrade/src/test/java/org/opensearch/upgrades/ClusterStatsIT.java b/qa/rolling-upgrade/src/test/java/org/opensearch/upgrades/ClusterStatsIT.java new file mode 100644 index 0000000000000..1c5f35db8ec46 --- /dev/null +++ b/qa/rolling-upgrade/src/test/java/org/opensearch/upgrades/ClusterStatsIT.java @@ -0,0 +1,67 @@ +/* + * SPDX-License-Identifier: Apache-2.0 + * + * The OpenSearch Contributors require contributions made to + * this file be licensed under the Apache-2.0 license or a + * compatible open source license. + */ + +package org.opensearch.upgrades; + +import org.opensearch.Version; +import org.opensearch.client.Request; +import org.opensearch.client.Response; + +import java.io.IOException; +import java.util.Collections; +import java.util.List; +import java.util.Map; + +public class ClusterStatsIT extends AbstractRollingTestCase { + + private final List nodeStatsMetrics = List.of("os", "process", "jvm", "fs", "plugins", "ingest", "network_types", "discovery_types", "packaging_types"); + + private final List indicesStatsMetrics = List.of("shards", "docs", "store", "fielddata", "query_cache", "completion", "segments", "analysis", "mappings"); + + public void testClusterStats() throws IOException { + Response response = client().performRequest(new Request("GET", "/_cluster/stats")); + validateClusterStatsWithFilterResponse(response, nodeStatsMetrics, indicesStatsMetrics); + if (AbstractRollingTestCase.UPGRADE_FROM_VERSION.onOrAfter(Version.V_3_0_0) || ( + CLUSTER_TYPE == ClusterType.UPGRADED && Version.CURRENT.onOrAfter(Version.V_3_0_0))) { + response = client().performRequest(new Request("GET", "/_cluster/stats/os/nodes/_all")); + validateClusterStatsWithFilterResponse(response, List.of("os"), Collections.emptyList()); + response = client().performRequest(new Request("GET", "/_cluster/stats/indices/mappings/nodes/_all")); + validateClusterStatsWithFilterResponse(response, Collections.emptyList(), List.of("mappings")); + response = client().performRequest(new Request("GET", "/_cluster/stats/os,indices/mappings/nodes/_all")); + validateClusterStatsWithFilterResponse(response, List.of("os"), List.of("mappings")); + } + } + + private void validateClusterStatsWithFilterResponse(Response response, List requestedNodesStatsMetrics, List requestedIndicesStatsMetrics) throws IOException { + assertEquals(200, response.getStatusLine().getStatusCode()); + Map entity = entityAsMap(response); + if (requestedNodesStatsMetrics != null && !requestedNodesStatsMetrics.isEmpty()) { + assertTrue(entity.containsKey("nodes")); + Map nodesStats = (Map) entity.get("nodes"); + for (String metric : nodeStatsMetrics) { + if (requestedNodesStatsMetrics.contains(metric)) { + assertTrue(nodesStats.containsKey(metric)); + } else { + assertFalse(nodesStats.containsKey(metric)); + } + } + } + + if (requestedIndicesStatsMetrics != null && !requestedIndicesStatsMetrics.isEmpty()) { + assertTrue(entity.containsKey("indices")); + Map indicesStats = (Map) entity.get("indices"); + for (String metric : indicesStatsMetrics) { + if (requestedIndicesStatsMetrics.contains(metric)) { + assertTrue(indicesStats.containsKey(metric)); + } else { + assertFalse(indicesStats.containsKey(metric)); + } + } + } + } +} diff --git a/server/src/internalClusterTest/java/org/opensearch/action/admin/cluster/stats/ClusterStatsIT.java b/server/src/internalClusterTest/java/org/opensearch/action/admin/cluster/stats/ClusterStatsIT.java index f23cdbb50b37a..5f00ba35c7b69 100644 --- a/server/src/internalClusterTest/java/org/opensearch/action/admin/cluster/stats/ClusterStatsIT.java +++ b/server/src/internalClusterTest/java/org/opensearch/action/admin/cluster/stats/ClusterStatsIT.java @@ -37,6 +37,9 @@ import org.opensearch.action.admin.cluster.node.stats.NodeStats; import org.opensearch.action.admin.cluster.node.stats.NodesStatsRequest; import org.opensearch.action.admin.cluster.node.stats.NodesStatsResponse; +import org.opensearch.action.admin.cluster.stats.ClusterStatsRequest.IndexMetric; +import org.opensearch.action.admin.cluster.stats.ClusterStatsRequest.Metric; +import org.opensearch.action.index.IndexRequest; import org.opensearch.client.Client; import org.opensearch.client.Requests; import org.opensearch.cluster.health.ClusterHealthStatus; @@ -44,6 +47,7 @@ import org.opensearch.common.Priority; import org.opensearch.common.settings.Settings; import org.opensearch.common.util.concurrent.OpenSearchExecutors; +import org.opensearch.core.common.unit.ByteSizeValue; import org.opensearch.monitor.os.OsStats; import org.opensearch.node.NodeRoleSettings; import org.opensearch.test.OpenSearchIntegTestCase; @@ -230,6 +234,7 @@ public void testIndicesShardStatsWithoutNodeLevelAggregations() { } public void testIndicesShardStatsWithNodeLevelAggregations() { + internalCluster().startNode(); ensureGreen(); ClusterStatsResponse response = client().admin().cluster().prepareClusterStats().useAggregatedNodeLevelResponses(true).get(); @@ -317,6 +322,53 @@ public void testValuesSmokeScreen() throws IOException, ExecutionException, Inte assertEquals(msg, OsStats.calculatePercentage(free, total), response.nodesStats.getOs().getMem().getFreePercent()); } + public void testValuesSmokeScreenWithNodeStatsAndIndicesStatsMetricsFilter() throws IOException, ExecutionException, + InterruptedException { + internalCluster().startNodes(randomIntBetween(1, 3)); + index("test1", "type", "1", "f", "f"); + + ClusterStatsResponse response = client().admin() + .cluster() + .prepareClusterStats() + .useAggregatedNodeLevelResponses(randomBoolean()) + .computeAllMetrics(false) + .requestMetrics(Set.of(Metric.values())) + .indexMetrics(Set.of(IndexMetric.values())) + .get(); + String msg = response.toString(); + assertThat(msg, response.getTimestamp(), Matchers.greaterThan(946681200000L)); // 1 Jan 2000 + assertThat(msg, response.indicesStats.getStore().getSizeInBytes(), Matchers.greaterThan(0L)); + + assertThat(msg, response.nodesStats.getFs().getTotal().getBytes(), Matchers.greaterThan(0L)); + assertThat(msg, response.nodesStats.getJvm().getVersions().size(), Matchers.greaterThan(0)); + + assertThat(msg, response.nodesStats.getVersions().size(), Matchers.greaterThan(0)); + assertThat(msg, response.nodesStats.getVersions().contains(Version.CURRENT), Matchers.equalTo(true)); + assertThat(msg, response.nodesStats.getPlugins().size(), Matchers.greaterThanOrEqualTo(0)); + + assertThat(msg, response.nodesStats.getProcess().count, Matchers.greaterThan(0)); + // 0 happens when not supported on platform + assertThat(msg, response.nodesStats.getProcess().getAvgOpenFileDescriptors(), Matchers.greaterThanOrEqualTo(0L)); + // these can be -1 if not supported on platform + assertThat(msg, response.nodesStats.getProcess().getMinOpenFileDescriptors(), Matchers.greaterThanOrEqualTo(-1L)); + assertThat(msg, response.nodesStats.getProcess().getMaxOpenFileDescriptors(), Matchers.greaterThanOrEqualTo(-1L)); + + NodesStatsResponse nodesStatsResponse = client().admin().cluster().prepareNodesStats().addMetric(OS.metricName()).get(); + long total = 0; + long free = 0; + long used = 0; + for (NodeStats nodeStats : nodesStatsResponse.getNodes()) { + total += nodeStats.getOs().getMem().getTotal().getBytes(); + free += nodeStats.getOs().getMem().getFree().getBytes(); + used += nodeStats.getOs().getMem().getUsed().getBytes(); + } + assertEquals(msg, free, response.nodesStats.getOs().getMem().getFree().getBytes()); + assertEquals(msg, total, response.nodesStats.getOs().getMem().getTotal().getBytes()); + assertEquals(msg, used, response.nodesStats.getOs().getMem().getUsed().getBytes()); + assertEquals(msg, OsStats.calculatePercentage(used, total), response.nodesStats.getOs().getMem().getUsedPercent()); + assertEquals(msg, OsStats.calculatePercentage(free, total), response.nodesStats.getOs().getMem().getFreePercent()); + } + public void testAllocatedProcessors() throws Exception { // start one node with 7 processors. internalCluster().startNode(Settings.builder().put(OpenSearchExecutors.NODE_PROCESSORS_SETTING.getKey(), 7).build()); @@ -384,6 +436,43 @@ public void testFieldTypes() { } } + public void testFieldTypesWithMappingsFilter() { + internalCluster().startNode(); + ensureGreen(); + ClusterStatsResponse response = client().admin() + .cluster() + .prepareClusterStats() + .useAggregatedNodeLevelResponses(randomBoolean()) + .computeAllMetrics(randomBoolean()) + .requestMetrics(Set.of(Metric.INDICES)) + .indexMetrics(Set.of(IndexMetric.MAPPINGS)) + .get(); + assertThat(response.getStatus(), Matchers.equalTo(ClusterHealthStatus.GREEN)); + assertTrue(response.getIndicesStats().getMappings().getFieldTypeStats().isEmpty()); + + client().admin().indices().prepareCreate("test1").setMapping("{\"properties\":{\"foo\":{\"type\": \"keyword\"}}}").get(); + client().admin() + .indices() + .prepareCreate("test2") + .setMapping( + "{\"properties\":{\"foo\":{\"type\": \"keyword\"},\"bar\":{\"properties\":{\"baz\":{\"type\":\"keyword\"}," + + "\"eggplant\":{\"type\":\"integer\"}}}}}" + ) + .get(); + response = client().admin().cluster().prepareClusterStats().useAggregatedNodeLevelResponses(randomBoolean()).get(); + assertThat(response.getIndicesStats().getMappings().getFieldTypeStats().size(), equalTo(3)); + Set stats = response.getIndicesStats().getMappings().getFieldTypeStats(); + for (IndexFeatureStats stat : stats) { + if (stat.getName().equals("integer")) { + assertThat(stat.getCount(), greaterThanOrEqualTo(1)); + } else if (stat.getName().equals("keyword")) { + assertThat(stat.getCount(), greaterThanOrEqualTo(3)); + } else if (stat.getName().equals("object")) { + assertThat(stat.getCount(), greaterThanOrEqualTo(1)); + } + } + } + public void testNodeRolesWithMasterLegacySettings() throws ExecutionException, InterruptedException { int total = 1; Settings legacyMasterSettings = Settings.builder() @@ -505,6 +594,293 @@ public void testNodeRolesWithDataNodeLegacySettings() throws ExecutionException, assertEquals(expectedNodesRoles, Set.of(getNodeRoles(client, 0), getNodeRoles(client, 1))); } + public void testClusterStatsWithNodeMetricsFilter() { + internalCluster().startNode(); + ensureGreen(); + + client().admin().indices().prepareCreate("test1").setMapping("{\"properties\":{\"foo\":{\"type\": \"keyword\"}}}").get(); + + ClusterStatsRequestBuilder clusterStatsRequestBuilder = client().admin() + .cluster() + .prepareClusterStats() + .useAggregatedNodeLevelResponses(randomBoolean()); + assertTrue(clusterStatsRequestBuilder.request().computeAllMetrics()); + + ClusterStatsResponse response = clusterStatsRequestBuilder.get(); + assertNotNull(response); + assertNotNull(response.getNodesStats()); + assertNotNull(response.getIndicesStats()); + + ClusterStatsResponse statsResponseWithAllNodeStatsMetrics = client().admin() + .cluster() + .prepareClusterStats() + .useAggregatedNodeLevelResponses(randomBoolean()) + .requestMetrics(ClusterStatsNodes.NODE_STATS_METRICS) + .computeAllMetrics(false) + .get(); + assertNotNull(statsResponseWithAllNodeStatsMetrics); + assertNotNull(statsResponseWithAllNodeStatsMetrics.getNodesStats()); + assertNull(statsResponseWithAllNodeStatsMetrics.getIndicesStats()); + validateNodeStatsOutput(ClusterStatsNodes.NODE_STATS_METRICS, statsResponseWithAllNodeStatsMetrics); + assertEquals( + response.getNodesStats().getCounts().getTotal(), + statsResponseWithAllNodeStatsMetrics.getNodesStats().getCounts().getTotal() + ); + assertEquals( + response.getNodesStats().getCounts().getRoles(), + statsResponseWithAllNodeStatsMetrics.getNodesStats().getCounts().getRoles() + ); + assertEquals(response.getNodesStats().getVersions(), statsResponseWithAllNodeStatsMetrics.getNodesStats().getVersions()); + assertEquals(response.getNodesStats().getPlugins(), statsResponseWithAllNodeStatsMetrics.getNodesStats().getPlugins()); + } + + public void testClusterStatsWithIndicesOnlyMetricsFilter() { + internalCluster().startNode(); + ensureGreen(); + + client().admin().indices().prepareCreate("test1").setMapping("{\"properties\":{\"foo\":{\"type\": \"keyword\"}}}").get(); + + ClusterStatsRequestBuilder clusterStatsRequestBuilder = client().admin() + .cluster() + .prepareClusterStats() + .useAggregatedNodeLevelResponses(randomBoolean()); + assertTrue(clusterStatsRequestBuilder.request().computeAllMetrics()); + + ClusterStatsResponse response = clusterStatsRequestBuilder.get(); + assertNotNull(response); + assertNotNull(response.getNodesStats()); + assertNotNull(response.getIndicesStats()); + + ClusterStatsResponse statsResponseWithIndicesRequestMetrics = client().admin() + .cluster() + .prepareClusterStats() + .useAggregatedNodeLevelResponses(randomBoolean()) + .requestMetrics(Set.of(Metric.INDICES)) + .indexMetrics(Set.of(IndexMetric.values())) + .computeAllMetrics(false) + .get(); + assertNotNull(statsResponseWithIndicesRequestMetrics); + assertNull(statsResponseWithIndicesRequestMetrics.getNodesStats()); + assertNotNull(statsResponseWithIndicesRequestMetrics.getIndicesStats()); + validateIndicesStatsOutput(Set.of(IndexMetric.values()), statsResponseWithIndicesRequestMetrics); + } + + public void testClusterStatsWithSelectiveNodeMetricAndIndexMetricsFilter() { + internalCluster().startNode(); + ensureGreen(); + + client().admin().indices().prepareCreate("test1").setMapping("{\"properties\":{\"foo\":{\"type\": \"keyword\"}}}").get(); + IndexRequest indexRequest = new IndexRequest("test1").id("doc_id").source(Map.of("test_type", "metrics_filter")); + client().index(indexRequest); + + ClusterStatsRequestBuilder clusterStatsRequestBuilder = client().admin() + .cluster() + .prepareClusterStats() + .useAggregatedNodeLevelResponses(randomBoolean()); + assertTrue(clusterStatsRequestBuilder.request().computeAllMetrics()); + + ClusterStatsResponse response = clusterStatsRequestBuilder.get(); + assertNotNull(response); + assertNotNull(response.getNodesStats()); + assertNotNull(response.getIndicesStats()); + + ClusterStatsResponse statsResponseWithAllIndicesMetrics = client().admin() + .cluster() + .prepareClusterStats() + .useAggregatedNodeLevelResponses(randomBoolean()) + .requestMetrics(Set.of(Metric.OS, Metric.FS, Metric.INDICES)) + .indexMetrics(Set.of(IndexMetric.FIELDDATA, IndexMetric.SHARDS, IndexMetric.SEGMENTS, IndexMetric.DOCS, IndexMetric.STORE)) + .computeAllMetrics(false) + .get(); + assertNotNull(statsResponseWithAllIndicesMetrics); + assertNotNull(statsResponseWithAllIndicesMetrics.getNodesStats()); + assertNotNull(statsResponseWithAllIndicesMetrics.getIndicesStats()); + validateNodeStatsOutput(Set.of(Metric.FS, Metric.OS), statsResponseWithAllIndicesMetrics); + validateIndicesStatsOutput( + Set.of(IndexMetric.FIELDDATA, IndexMetric.SHARDS, IndexMetric.SEGMENTS, IndexMetric.DOCS, IndexMetric.STORE), + statsResponseWithAllIndicesMetrics + ); + assertEquals(response.getIndicesStats().getFieldData(), statsResponseWithAllIndicesMetrics.getIndicesStats().getFieldData()); + assertEquals(response.getIndicesStats().getIndexCount(), statsResponseWithAllIndicesMetrics.getIndicesStats().getIndexCount()); + assertEquals( + response.getIndicesStats().getShards().getTotal(), + statsResponseWithAllIndicesMetrics.getIndicesStats().getShards().getTotal() + ); + assertEquals( + response.getIndicesStats().getShards().getPrimaries(), + statsResponseWithAllIndicesMetrics.getIndicesStats().getShards().getPrimaries() + ); + } + + public void testClusterStatsWithMappingsAndAnalysisStatsIndexMetricsFilter() { + internalCluster().startNode(); + ensureGreen(); + + client().admin().indices().prepareCreate("test1").setMapping("{\"properties\":{\"foo\":{\"type\": \"keyword\"}}}").get(); + IndexRequest indexRequest = new IndexRequest("test1").id("doc_id").source(Map.of("test_type", "metrics_filter")); + client().index(indexRequest); + + ClusterStatsRequestBuilder clusterStatsRequestBuilder = client().admin() + .cluster() + .prepareClusterStats() + .useAggregatedNodeLevelResponses(randomBoolean()); + assertTrue(clusterStatsRequestBuilder.request().computeAllMetrics()); + + ClusterStatsResponse response = clusterStatsRequestBuilder.get(); + assertNotNull(response); + assertNotNull(response.getNodesStats()); + assertNotNull(response.getIndicesStats()); + + ClusterStatsResponse statsResponseWithSpecificIndicesMetrics = client().admin() + .cluster() + .prepareClusterStats() + .useAggregatedNodeLevelResponses(randomBoolean()) + .requestMetrics(Set.of(Metric.INDICES)) + .indexMetrics(Set.of(IndexMetric.MAPPINGS, IndexMetric.ANALYSIS)) + .computeAllMetrics(false) + .get(); + assertNotNull(statsResponseWithSpecificIndicesMetrics); + assertNull(statsResponseWithSpecificIndicesMetrics.getNodesStats()); + assertNotNull(statsResponseWithSpecificIndicesMetrics.getIndicesStats()); + validateIndicesStatsOutput(Set.of(IndexMetric.MAPPINGS, IndexMetric.ANALYSIS), statsResponseWithSpecificIndicesMetrics); + assertEquals(response.getIndicesStats().getIndexCount(), statsResponseWithSpecificIndicesMetrics.getIndicesStats().getIndexCount()); + assertEquals(response.getIndicesStats().getMappings(), statsResponseWithSpecificIndicesMetrics.getIndicesStats().getMappings()); + assertEquals(response.getIndicesStats().getAnalysis(), statsResponseWithSpecificIndicesMetrics.getIndicesStats().getAnalysis()); + } + + public void testClusterStatsWithIndexMetricWithDocsFilter() throws IOException { + internalCluster().startNode(); + createIndex("test1"); + + client().prepareIndex("test1").setId(Integer.toString(1)).setSource("field1", "value1").execute().actionGet(); + client().prepareIndex("test1").setId(Integer.toString(2)).setSource("field2", "value2").execute().actionGet(); + refreshAndWaitForReplication(); + + ClusterStatsResponse statsResponseWithAllIndicesMetrics = client().admin() + .cluster() + .prepareClusterStats() + .useAggregatedNodeLevelResponses(randomBoolean()) + .requestMetrics(Set.of(Metric.INDICES)) + .indexMetrics(Set.of(IndexMetric.DOCS)) + .computeAllMetrics(false) + .get(); + assertNotNull(statsResponseWithAllIndicesMetrics); + assertNull(statsResponseWithAllIndicesMetrics.getNodesStats()); + assertNotNull(statsResponseWithAllIndicesMetrics.getIndicesStats()); + validateIndicesStatsOutput(Set.of(IndexMetric.DOCS), statsResponseWithAllIndicesMetrics); + assertEquals(2, statsResponseWithAllIndicesMetrics.getIndicesStats().getDocs().getCount()); + assertEquals(0, statsResponseWithAllIndicesMetrics.getIndicesStats().getDocs().getDeleted()); + assertTrue(statsResponseWithAllIndicesMetrics.getIndicesStats().getDocs().getAverageSizeInBytes() > 0); + } + + public void testClusterStatsWithSelectiveMetricsFilterAndNoIndex() { + internalCluster().startNode(); + ensureGreen(); + ClusterStatsResponse statsResponseWithAllIndicesMetrics = client().admin() + .cluster() + .prepareClusterStats() + .useAggregatedNodeLevelResponses(randomBoolean()) + .requestMetrics(Set.of(Metric.OS, Metric.FS, Metric.INDICES)) + .indexMetrics(Set.of(IndexMetric.FIELDDATA, IndexMetric.SHARDS, IndexMetric.SEGMENTS, IndexMetric.DOCS, IndexMetric.STORE)) + .computeAllMetrics(false) + .get(); + assertNotNull(statsResponseWithAllIndicesMetrics); + assertNotNull(statsResponseWithAllIndicesMetrics.getNodesStats()); + assertNotNull(statsResponseWithAllIndicesMetrics.getIndicesStats()); + validateNodeStatsOutput(Set.of(Metric.FS, Metric.OS), statsResponseWithAllIndicesMetrics); + validateIndicesStatsOutput( + Set.of(IndexMetric.FIELDDATA, IndexMetric.SHARDS, IndexMetric.SEGMENTS, IndexMetric.DOCS, IndexMetric.STORE), + statsResponseWithAllIndicesMetrics + ); + assertEquals(0, statsResponseWithAllIndicesMetrics.getIndicesStats().getShards().getIndices()); + assertEquals(0, statsResponseWithAllIndicesMetrics.getIndicesStats().getShards().getTotal()); + assertEquals(0, statsResponseWithAllIndicesMetrics.getIndicesStats().getShards().getPrimaries()); + assertEquals(0, statsResponseWithAllIndicesMetrics.getIndicesStats().getDocs().getCount()); + assertEquals(0, statsResponseWithAllIndicesMetrics.getIndicesStats().getDocs().getDeleted()); + assertEquals(0, statsResponseWithAllIndicesMetrics.getIndicesStats().getDocs().getTotalSizeInBytes()); + assertEquals(0, statsResponseWithAllIndicesMetrics.getIndicesStats().getStore().getSizeInBytes()); + assertEquals(new ByteSizeValue(0), statsResponseWithAllIndicesMetrics.getIndicesStats().getStore().getReservedSize()); + assertEquals(new ByteSizeValue(0), statsResponseWithAllIndicesMetrics.getIndicesStats().getFieldData().getMemorySize()); + assertEquals(0, statsResponseWithAllIndicesMetrics.getIndicesStats().getFieldData().getEvictions()); + assertNull(statsResponseWithAllIndicesMetrics.getIndicesStats().getFieldData().getFields()); + assertEquals(0, statsResponseWithAllIndicesMetrics.getIndicesStats().getSegments().getCount()); + assertEquals(0, statsResponseWithAllIndicesMetrics.getIndicesStats().getSegments().getIndexWriterMemoryInBytes()); + assertEquals(0, statsResponseWithAllIndicesMetrics.getIndicesStats().getSegments().getVersionMapMemoryInBytes()); + } + + private void validateNodeStatsOutput(Set expectedMetrics, ClusterStatsResponse clusterStatsResponse) { + // Ingest, network types, discovery types and packaging types stats are not included here as they don't have a get method exposed. + Set NodeMetrics = Set.of(Metric.OS, Metric.JVM, Metric.FS, Metric.PROCESS, Metric.PLUGINS); + for (Metric metric : NodeMetrics) { + Object object = null; + switch (metric) { + case OS: + object = clusterStatsResponse.getNodesStats().getOs(); + break; + case JVM: + object = clusterStatsResponse.getNodesStats().getJvm(); + break; + case FS: + object = clusterStatsResponse.getNodesStats().getFs(); + break; + case PROCESS: + object = clusterStatsResponse.getNodesStats().getProcess(); + break; + case PLUGINS: + object = clusterStatsResponse.getNodesStats().getPlugins(); + break; + } + if (expectedMetrics.contains(metric)) { + assertNotNull(object); + } else { + assertNull(object); + } + } + } + + private void validateIndicesStatsOutput( + Set expectedMetrics, + ClusterStatsResponse clusterStatsResponse + ) { + for (IndexMetric indexMetric : IndexMetric.values()) { + Object object = null; + switch (indexMetric) { + case SHARDS: + object = clusterStatsResponse.getIndicesStats().getShards(); + break; + case DOCS: + object = clusterStatsResponse.getIndicesStats().getDocs(); + break; + case STORE: + object = clusterStatsResponse.getIndicesStats().getStore(); + break; + case FIELDDATA: + object = clusterStatsResponse.getIndicesStats().getFieldData(); + break; + case QUERY_CACHE: + object = clusterStatsResponse.getIndicesStats().getQueryCache(); + break; + case COMPLETION: + object = clusterStatsResponse.getIndicesStats().getCompletion(); + break; + case SEGMENTS: + object = clusterStatsResponse.getIndicesStats().getSegments(); + break; + case ANALYSIS: + object = clusterStatsResponse.getIndicesStats().getAnalysis(); + break; + case MAPPINGS: + object = clusterStatsResponse.getIndicesStats().getMappings(); + break; + } + if (expectedMetrics.contains(indexMetric)) { + assertNotNull(object); + } else { + assertNull(object); + } + } + } + private Map getExpectedCounts( int dataRoleCount, int masterRoleCount, diff --git a/server/src/internalClusterTest/java/org/opensearch/indices/settings/SearchOnlyReplicaIT.java b/server/src/internalClusterTest/java/org/opensearch/indices/settings/SearchOnlyReplicaIT.java index 6bd91df1de66f..fa836e2cc5784 100644 --- a/server/src/internalClusterTest/java/org/opensearch/indices/settings/SearchOnlyReplicaIT.java +++ b/server/src/internalClusterTest/java/org/opensearch/indices/settings/SearchOnlyReplicaIT.java @@ -126,6 +126,8 @@ public void testFailoverWithSearchReplica_WithoutWriterReplicas() throws IOExcep .put(indexSettings()) .put(IndexMetadata.SETTING_NUMBER_OF_REPLICAS, numWriterReplicas) .put(IndexMetadata.SETTING_NUMBER_OF_SEARCH_REPLICAS, numSearchReplicas) + .put("index.refresh_interval", "40ms") // set lower interval so replica attempts replication cycles after primary is + // removed. .build() ); ensureYellow(TEST_INDEX); diff --git a/server/src/main/java/org/opensearch/action/admin/cluster/stats/ClusterStatsIndices.java b/server/src/main/java/org/opensearch/action/admin/cluster/stats/ClusterStatsIndices.java index 03a73f45ffe81..9ebe36531c208 100644 --- a/server/src/main/java/org/opensearch/action/admin/cluster/stats/ClusterStatsIndices.java +++ b/server/src/main/java/org/opensearch/action/admin/cluster/stats/ClusterStatsIndices.java @@ -32,6 +32,7 @@ package org.opensearch.action.admin.cluster.stats; +import org.opensearch.action.admin.cluster.stats.ClusterStatsRequest.IndexMetric; import org.opensearch.action.admin.indices.stats.CommonStats; import org.opensearch.common.annotation.PublicApi; import org.opensearch.core.xcontent.ToXContentFragment; @@ -47,6 +48,7 @@ import java.util.HashMap; import java.util.List; import java.util.Map; +import java.util.Set; /** * Cluster Stats per index @@ -68,14 +70,23 @@ public class ClusterStatsIndices implements ToXContentFragment { private MappingStats mappings; public ClusterStatsIndices(List nodeResponses, MappingStats mappingStats, AnalysisStats analysisStats) { - Map countsPerIndex = new HashMap<>(); + this(Set.of(IndexMetric.values()), nodeResponses, mappingStats, analysisStats); + + } - this.docs = new DocsStats(); - this.store = new StoreStats(); - this.fieldData = new FieldDataStats(); - this.queryCache = new QueryCacheStats(); - this.completion = new CompletionStats(); - this.segments = new SegmentsStats(); + public ClusterStatsIndices( + Set indicesMetrics, + List nodeResponses, + MappingStats mappingStats, + AnalysisStats analysisStats + ) { + Map countsPerIndex = new HashMap<>(); + this.docs = indicesMetrics.contains(IndexMetric.DOCS) ? new DocsStats() : null; + this.store = indicesMetrics.contains(IndexMetric.STORE) ? new StoreStats() : null; + this.fieldData = indicesMetrics.contains(IndexMetric.FIELDDATA) ? new FieldDataStats() : null; + this.queryCache = indicesMetrics.contains(IndexMetric.QUERY_CACHE) ? new QueryCacheStats() : null; + this.completion = indicesMetrics.contains(IndexMetric.COMPLETION) ? new CompletionStats() : null; + this.segments = indicesMetrics.contains(IndexMetric.SEGMENTS) ? new SegmentsStats() : null; for (ClusterStatsNodeResponse r : nodeResponses) { // Aggregated response from the node @@ -92,12 +103,24 @@ public ClusterStatsIndices(List nodeResponses, Mapping } } - docs.add(r.getAggregatedNodeLevelStats().commonStats.docs); - store.add(r.getAggregatedNodeLevelStats().commonStats.store); - fieldData.add(r.getAggregatedNodeLevelStats().commonStats.fieldData); - queryCache.add(r.getAggregatedNodeLevelStats().commonStats.queryCache); - completion.add(r.getAggregatedNodeLevelStats().commonStats.completion); - segments.add(r.getAggregatedNodeLevelStats().commonStats.segments); + if (indicesMetrics.contains(IndexMetric.DOCS)) { + docs.add(r.getAggregatedNodeLevelStats().commonStats.docs); + } + if (indicesMetrics.contains(IndexMetric.STORE)) { + store.add(r.getAggregatedNodeLevelStats().commonStats.store); + } + if (indicesMetrics.contains(IndexMetric.FIELDDATA)) { + fieldData.add(r.getAggregatedNodeLevelStats().commonStats.fieldData); + } + if (indicesMetrics.contains(IndexMetric.QUERY_CACHE)) { + queryCache.add(r.getAggregatedNodeLevelStats().commonStats.queryCache); + } + if (indicesMetrics.contains(IndexMetric.COMPLETION)) { + completion.add(r.getAggregatedNodeLevelStats().commonStats.completion); + } + if (indicesMetrics.contains(IndexMetric.SEGMENTS)) { + segments.add(r.getAggregatedNodeLevelStats().commonStats.segments); + } } else { // Default response from the node for (org.opensearch.action.admin.indices.stats.ShardStats shardStats : r.shardsStats()) { @@ -113,21 +136,35 @@ public ClusterStatsIndices(List nodeResponses, Mapping if (shardStats.getShardRouting().primary()) { indexShardStats.primaries++; - docs.add(shardCommonStats.docs); + if (indicesMetrics.contains(IndexMetric.DOCS)) { + docs.add(shardCommonStats.docs); + } + } + if (indicesMetrics.contains(IndexMetric.STORE)) { + store.add(shardCommonStats.store); + } + if (indicesMetrics.contains(IndexMetric.FIELDDATA)) { + fieldData.add(shardCommonStats.fieldData); + } + if (indicesMetrics.contains(IndexMetric.QUERY_CACHE)) { + queryCache.add(shardCommonStats.queryCache); + } + if (indicesMetrics.contains(IndexMetric.COMPLETION)) { + completion.add(shardCommonStats.completion); + } + if (indicesMetrics.contains(IndexMetric.SEGMENTS)) { + segments.add(shardCommonStats.segments); } - store.add(shardCommonStats.store); - fieldData.add(shardCommonStats.fieldData); - queryCache.add(shardCommonStats.queryCache); - completion.add(shardCommonStats.completion); - segments.add(shardCommonStats.segments); } } } - shards = new ShardStats(); indexCount = countsPerIndex.size(); - for (final ShardStats indexCountsCursor : countsPerIndex.values()) { - shards.addIndexShardCount(indexCountsCursor); + if (indicesMetrics.contains(IndexMetric.SHARDS)) { + shards = new ShardStats(); + for (final ShardStats indexCountsCursor : countsPerIndex.values()) { + shards.addIndexShardCount(indexCountsCursor); + } } this.mappings = mappingStats; @@ -186,13 +223,27 @@ static final class Fields { @Override public XContentBuilder toXContent(XContentBuilder builder, Params params) throws IOException { builder.field(Fields.COUNT, indexCount); - shards.toXContent(builder, params); - docs.toXContent(builder, params); - store.toXContent(builder, params); - fieldData.toXContent(builder, params); - queryCache.toXContent(builder, params); - completion.toXContent(builder, params); - segments.toXContent(builder, params); + if (shards != null) { + shards.toXContent(builder, params); + } + if (docs != null) { + docs.toXContent(builder, params); + } + if (store != null) { + store.toXContent(builder, params); + } + if (fieldData != null) { + fieldData.toXContent(builder, params); + } + if (queryCache != null) { + queryCache.toXContent(builder, params); + } + if (completion != null) { + completion.toXContent(builder, params); + } + if (segments != null) { + segments.toXContent(builder, params); + } if (mappings != null) { mappings.toXContent(builder, params); } diff --git a/server/src/main/java/org/opensearch/action/admin/cluster/stats/ClusterStatsNodes.java b/server/src/main/java/org/opensearch/action/admin/cluster/stats/ClusterStatsNodes.java index b44e9cfc5c74a..bf8218a66fc17 100644 --- a/server/src/main/java/org/opensearch/action/admin/cluster/stats/ClusterStatsNodes.java +++ b/server/src/main/java/org/opensearch/action/admin/cluster/stats/ClusterStatsNodes.java @@ -36,6 +36,7 @@ import org.opensearch.action.admin.cluster.node.info.NodeInfo; import org.opensearch.action.admin.cluster.node.info.PluginsAndModules; import org.opensearch.action.admin.cluster.node.stats.NodeStats; +import org.opensearch.action.admin.cluster.stats.ClusterStatsRequest.Metric; import org.opensearch.cluster.node.DiscoveryNode; import org.opensearch.cluster.node.DiscoveryNodeRole; import org.opensearch.common.annotation.PublicApi; @@ -89,10 +90,29 @@ public class ClusterStatsNodes implements ToXContentFragment { private final PackagingTypes packagingTypes; private final IngestStats ingestStats; + public static final Set NODE_STATS_METRICS = Set.of( + // Stats computed from node info and node stat + Metric.OS, + Metric.JVM, + // Stats computed from node stat + Metric.FS, + Metric.PROCESS, + Metric.INGEST, + // Stats computed from node info + Metric.PLUGINS, + Metric.NETWORK_TYPES, + Metric.DISCOVERY_TYPES, + Metric.PACKAGING_TYPES + ); + ClusterStatsNodes(List nodeResponses) { + this(Set.of(Metric.values()), nodeResponses); + } + + ClusterStatsNodes(Set requestedMetrics, List nodeResponses) { this.versions = new HashSet<>(); - this.fs = new FsInfo.Path(); - this.plugins = new HashSet<>(); + this.fs = requestedMetrics.contains(ClusterStatsRequest.Metric.FS) ? new FsInfo.Path() : null; + this.plugins = requestedMetrics.contains(ClusterStatsRequest.Metric.PLUGINS) ? new HashSet<>() : null; Set seenAddresses = new HashSet<>(nodeResponses.size()); List nodeInfos = new ArrayList<>(nodeResponses.size()); @@ -101,7 +121,9 @@ public class ClusterStatsNodes implements ToXContentFragment { nodeInfos.add(nodeResponse.nodeInfo()); nodeStats.add(nodeResponse.nodeStats()); this.versions.add(nodeResponse.nodeInfo().getVersion()); - this.plugins.addAll(nodeResponse.nodeInfo().getInfo(PluginsAndModules.class).getPluginInfos()); + if (requestedMetrics.contains(ClusterStatsRequest.Metric.PLUGINS)) { + this.plugins.addAll(nodeResponse.nodeInfo().getInfo(PluginsAndModules.class).getPluginInfos()); + } // now do the stats that should be deduped by hardware (implemented by ip deduping) TransportAddress publishAddress = nodeResponse.nodeInfo().getInfo(TransportInfo.class).address().publishAddress(); @@ -109,18 +131,19 @@ public class ClusterStatsNodes implements ToXContentFragment { if (!seenAddresses.add(inetAddress)) { continue; } - if (nodeResponse.nodeStats().getFs() != null) { + if (requestedMetrics.contains(ClusterStatsRequest.Metric.FS) && nodeResponse.nodeStats().getFs() != null) { this.fs.add(nodeResponse.nodeStats().getFs().getTotal()); } } + this.counts = new Counts(nodeInfos); - this.os = new OsStats(nodeInfos, nodeStats); - this.process = new ProcessStats(nodeStats); - this.jvm = new JvmStats(nodeInfos, nodeStats); - this.networkTypes = new NetworkTypes(nodeInfos); - this.discoveryTypes = new DiscoveryTypes(nodeInfos); - this.packagingTypes = new PackagingTypes(nodeInfos); - this.ingestStats = new IngestStats(nodeStats); + this.networkTypes = requestedMetrics.contains(ClusterStatsRequest.Metric.NETWORK_TYPES) ? new NetworkTypes(nodeInfos) : null; + this.discoveryTypes = requestedMetrics.contains(ClusterStatsRequest.Metric.DISCOVERY_TYPES) ? new DiscoveryTypes(nodeInfos) : null; + this.packagingTypes = requestedMetrics.contains(ClusterStatsRequest.Metric.PACKAGING_TYPES) ? new PackagingTypes(nodeInfos) : null; + this.ingestStats = requestedMetrics.contains(ClusterStatsRequest.Metric.INGEST) ? new IngestStats(nodeStats) : null; + this.process = requestedMetrics.contains(ClusterStatsRequest.Metric.PROCESS) ? new ProcessStats(nodeStats) : null; + this.os = requestedMetrics.contains(ClusterStatsRequest.Metric.OS) ? new OsStats(nodeInfos, nodeStats) : null; + this.jvm = requestedMetrics.contains(ClusterStatsRequest.Metric.JVM) ? new JvmStats(nodeInfos, nodeStats) : null; } public Counts getCounts() { @@ -179,36 +202,54 @@ public XContentBuilder toXContent(XContentBuilder builder, Params params) throws } builder.endArray(); - builder.startObject(Fields.OS); - os.toXContent(builder, params); - builder.endObject(); + if (os != null) { + builder.startObject(Fields.OS); + os.toXContent(builder, params); + builder.endObject(); + } - builder.startObject(Fields.PROCESS); - process.toXContent(builder, params); - builder.endObject(); + if (process != null) { + builder.startObject(Fields.PROCESS); + process.toXContent(builder, params); + builder.endObject(); + } - builder.startObject(Fields.JVM); - jvm.toXContent(builder, params); - builder.endObject(); + if (jvm != null) { + builder.startObject(Fields.JVM); + jvm.toXContent(builder, params); + builder.endObject(); + } - builder.field(Fields.FS); - fs.toXContent(builder, params); + if (fs != null) { + builder.field(Fields.FS); + fs.toXContent(builder, params); + } - builder.startArray(Fields.PLUGINS); - for (PluginInfo pluginInfo : plugins) { - pluginInfo.toXContent(builder, params); + if (plugins != null) { + builder.startArray(Fields.PLUGINS); + for (PluginInfo pluginInfo : plugins) { + pluginInfo.toXContent(builder, params); + } + builder.endArray(); } - builder.endArray(); - builder.startObject(Fields.NETWORK_TYPES); - networkTypes.toXContent(builder, params); - builder.endObject(); + if (networkTypes != null) { + builder.startObject(Fields.NETWORK_TYPES); + networkTypes.toXContent(builder, params); + builder.endObject(); + } - discoveryTypes.toXContent(builder, params); + if (discoveryTypes != null) { + discoveryTypes.toXContent(builder, params); + } - packagingTypes.toXContent(builder, params); + if (packagingTypes != null) { + packagingTypes.toXContent(builder, params); + } - ingestStats.toXContent(builder, params); + if (ingestStats != null) { + ingestStats.toXContent(builder, params); + } return builder; } diff --git a/server/src/main/java/org/opensearch/action/admin/cluster/stats/ClusterStatsRequest.java b/server/src/main/java/org/opensearch/action/admin/cluster/stats/ClusterStatsRequest.java index b82a9d256a134..1c929881b898b 100644 --- a/server/src/main/java/org/opensearch/action/admin/cluster/stats/ClusterStatsRequest.java +++ b/server/src/main/java/org/opensearch/action/admin/cluster/stats/ClusterStatsRequest.java @@ -39,6 +39,8 @@ import org.opensearch.core.common.io.stream.StreamOutput; import java.io.IOException; +import java.util.HashSet; +import java.util.Set; /** * A request to get cluster level stats. @@ -48,11 +50,30 @@ @PublicApi(since = "1.0.0") public class ClusterStatsRequest extends BaseNodesRequest { + private final Set requestedMetrics = new HashSet<>(); + private final Set indexMetricsRequested = new HashSet<>(); + private Boolean computeAllMetrics = true; + public ClusterStatsRequest(StreamInput in) throws IOException { super(in); if (in.getVersion().onOrAfter(Version.V_2_16_0)) { useAggregatedNodeLevelResponses = in.readOptionalBoolean(); } + if (in.getVersion().onOrAfter(Version.V_3_0_0)) { + computeAllMetrics = in.readOptionalBoolean(); + final long longMetricsFlags = in.readLong(); + for (Metric metric : Metric.values()) { + if ((longMetricsFlags & (1 << metric.getIndex())) != 0) { + requestedMetrics.add(metric); + } + } + final long longIndexMetricFlags = in.readLong(); + for (IndexMetric indexMetric : IndexMetric.values()) { + if ((longIndexMetricFlags & (1 << indexMetric.getIndex())) != 0) { + indexMetricsRequested.add(indexMetric); + } + } + } } private Boolean useAggregatedNodeLevelResponses = false; @@ -73,12 +94,133 @@ public void useAggregatedNodeLevelResponses(boolean useAggregatedNodeLevelRespon this.useAggregatedNodeLevelResponses = useAggregatedNodeLevelResponses; } + public boolean computeAllMetrics() { + return computeAllMetrics; + } + + public void computeAllMetrics(boolean computeAllMetrics) { + this.computeAllMetrics = computeAllMetrics; + } + + /** + * Add Metric + */ + public ClusterStatsRequest addMetric(Metric metric) { + requestedMetrics.add(metric); + return this; + } + + /** + * Get the names of requested metrics + */ + public Set requestedMetrics() { + return new HashSet<>(requestedMetrics); + } + + /** + * Add IndexMetric + */ + public ClusterStatsRequest addIndexMetric(IndexMetric indexMetric) { + indexMetricsRequested.add(indexMetric); + return this; + } + + public Set indicesMetrics() { + return new HashSet<>(indexMetricsRequested); + } + @Override public void writeTo(StreamOutput out) throws IOException { super.writeTo(out); if (out.getVersion().onOrAfter(Version.V_2_16_0)) { out.writeOptionalBoolean(useAggregatedNodeLevelResponses); } + if (out.getVersion().onOrAfter(Version.V_3_0_0)) { + out.writeOptionalBoolean(computeAllMetrics); + long longMetricFlags = 0; + for (Metric metric : requestedMetrics) { + longMetricFlags |= (1 << metric.getIndex()); + } + out.writeLong(longMetricFlags); + long longIndexMetricFlags = 0; + for (IndexMetric indexMetric : indexMetricsRequested) { + longIndexMetricFlags |= (1 << indexMetric.getIndex()); + } + out.writeLong(longIndexMetricFlags); + } } + /** + * An enumeration of the "core" sections of metrics that may be requested + * from the cluster stats endpoint. + */ + @PublicApi(since = "3.0.0") + public enum Metric { + OS("os", 0), + JVM("jvm", 1), + FS("fs", 2), + PROCESS("process", 3), + INGEST("ingest", 4), + PLUGINS("plugins", 5), + NETWORK_TYPES("network_types", 6), + DISCOVERY_TYPES("discovery_types", 7), + PACKAGING_TYPES("packaging_types", 8), + INDICES("indices", 9); + + private String metricName; + + private int index; + + Metric(String name, int index) { + this.metricName = name; + this.index = index; + } + + public String metricName() { + return this.metricName; + } + + public int getIndex() { + return index; + } + + } + + /** + * An enumeration of the "core" sections of indices metrics that may be requested + * from the cluster stats endpoint. + * + * When no value is provided for param index_metric, default filter is set to _all. + */ + @PublicApi(since = "3.0.0") + public enum IndexMetric { + // Metrics computed from ShardStats + SHARDS("shards", 0), + DOCS("docs", 1), + STORE("store", 2), + FIELDDATA("fielddata", 3), + QUERY_CACHE("query_cache", 4), + COMPLETION("completion", 5), + SEGMENTS("segments", 6), + // Metrics computed from ClusterState + ANALYSIS("analysis", 7), + MAPPINGS("mappings", 8); + + private String metricName; + + private int index; + + IndexMetric(String name, int index) { + this.metricName = name; + this.index = index; + } + + public String metricName() { + return this.metricName; + } + + public int getIndex() { + return this.index; + } + } } diff --git a/server/src/main/java/org/opensearch/action/admin/cluster/stats/ClusterStatsRequestBuilder.java b/server/src/main/java/org/opensearch/action/admin/cluster/stats/ClusterStatsRequestBuilder.java index 4d0932bd3927d..34fd9ea06235e 100644 --- a/server/src/main/java/org/opensearch/action/admin/cluster/stats/ClusterStatsRequestBuilder.java +++ b/server/src/main/java/org/opensearch/action/admin/cluster/stats/ClusterStatsRequestBuilder.java @@ -36,6 +36,11 @@ import org.opensearch.client.OpenSearchClient; import org.opensearch.common.annotation.PublicApi; +import java.util.Set; + +import static org.opensearch.action.admin.cluster.stats.ClusterStatsRequest.IndexMetric; +import static org.opensearch.action.admin.cluster.stats.ClusterStatsRequest.Metric; + /** * Transport request builder for obtaining cluster stats * @@ -55,4 +60,19 @@ public final ClusterStatsRequestBuilder useAggregatedNodeLevelResponses(boolean request.useAggregatedNodeLevelResponses(useAggregatedNodeLevelResponses); return this; } + + public final ClusterStatsRequestBuilder computeAllMetrics(boolean applyMetricFiltering) { + request.computeAllMetrics(applyMetricFiltering); + return this; + } + + public final ClusterStatsRequestBuilder requestMetrics(Set requestMetrics) { + requestMetrics.forEach(request::addMetric); + return this; + } + + public final ClusterStatsRequestBuilder indexMetrics(Set indexMetrics) { + indexMetrics.forEach(request::addIndexMetric); + return this; + } } diff --git a/server/src/main/java/org/opensearch/action/admin/cluster/stats/ClusterStatsResponse.java b/server/src/main/java/org/opensearch/action/admin/cluster/stats/ClusterStatsResponse.java index cc002b689a2a5..870996bb61b23 100644 --- a/server/src/main/java/org/opensearch/action/admin/cluster/stats/ClusterStatsResponse.java +++ b/server/src/main/java/org/opensearch/action/admin/cluster/stats/ClusterStatsResponse.java @@ -33,6 +33,8 @@ package org.opensearch.action.admin.cluster.stats; import org.opensearch.action.FailedNodeException; +import org.opensearch.action.admin.cluster.stats.ClusterStatsRequest.IndexMetric; +import org.opensearch.action.admin.cluster.stats.ClusterStatsRequest.Metric; import org.opensearch.action.support.nodes.BaseNodesResponse; import org.opensearch.cluster.ClusterName; import org.opensearch.cluster.ClusterState; @@ -47,6 +49,7 @@ import java.io.IOException; import java.util.List; import java.util.Locale; +import java.util.Set; /** * Transport response for obtaining cluster stats @@ -88,12 +91,31 @@ public ClusterStatsResponse( List nodes, List failures, ClusterState state + ) { + this(timestamp, clusterUUID, clusterName, nodes, failures, state, Set.of(Metric.values()), Set.of(IndexMetric.values())); + } + + public ClusterStatsResponse( + long timestamp, + String clusterUUID, + ClusterName clusterName, + List nodes, + List failures, + ClusterState state, + Set requestedMetrics, + Set indicesMetrics ) { super(clusterName, nodes, failures); this.clusterUUID = clusterUUID; this.timestamp = timestamp; - nodesStats = new ClusterStatsNodes(nodes); - indicesStats = new ClusterStatsIndices(nodes, MappingStats.of(state), AnalysisStats.of(state)); + nodesStats = requestedMetrics.stream().anyMatch(ClusterStatsNodes.NODE_STATS_METRICS::contains) + ? new ClusterStatsNodes(requestedMetrics, nodes) + : null; + MappingStats mappingStats = indicesMetrics.contains(IndexMetric.MAPPINGS) ? MappingStats.of(state) : null; + AnalysisStats analysisStats = indicesMetrics.contains(IndexMetric.ANALYSIS) ? AnalysisStats.of(state) : null; + indicesStats = requestedMetrics.contains(Metric.INDICES) + ? new ClusterStatsIndices(indicesMetrics, nodes, mappingStats, analysisStats) + : null; ClusterHealthStatus status = null; for (ClusterStatsNodeResponse response : nodes) { // only the cluster-manager node populates the status @@ -131,8 +153,13 @@ public void writeTo(StreamOutput out) throws IOException { out.writeVLong(timestamp); out.writeOptionalWriteable(status); out.writeOptionalString(clusterUUID); - out.writeOptionalWriteable(indicesStats.getMappings()); - out.writeOptionalWriteable(indicesStats.getAnalysis()); + if (indicesStats != null) { + out.writeOptionalWriteable(indicesStats.getMappings()); + out.writeOptionalWriteable(indicesStats.getAnalysis()); + } else { + out.writeOptionalWriteable(null); + out.writeOptionalWriteable(null); + } } @Override @@ -153,12 +180,16 @@ public XContentBuilder toXContent(XContentBuilder builder, Params params) throws if (status != null) { builder.field("status", status.name().toLowerCase(Locale.ROOT)); } - builder.startObject("indices"); - indicesStats.toXContent(builder, params); - builder.endObject(); - builder.startObject("nodes"); - nodesStats.toXContent(builder, params); - builder.endObject(); + if (indicesStats != null) { + builder.startObject("indices"); + indicesStats.toXContent(builder, params); + builder.endObject(); + } + if (nodesStats != null) { + builder.startObject("nodes"); + nodesStats.toXContent(builder, params); + builder.endObject(); + } return builder; } diff --git a/server/src/main/java/org/opensearch/action/admin/cluster/stats/TransportClusterStatsAction.java b/server/src/main/java/org/opensearch/action/admin/cluster/stats/TransportClusterStatsAction.java index c4b3524cf6da5..c6581b99eb559 100644 --- a/server/src/main/java/org/opensearch/action/admin/cluster/stats/TransportClusterStatsAction.java +++ b/server/src/main/java/org/opensearch/action/admin/cluster/stats/TransportClusterStatsAction.java @@ -37,6 +37,7 @@ import org.opensearch.action.admin.cluster.health.ClusterHealthRequest; import org.opensearch.action.admin.cluster.node.info.NodeInfo; import org.opensearch.action.admin.cluster.node.stats.NodeStats; +import org.opensearch.action.admin.cluster.stats.ClusterStatsRequest.Metric; import org.opensearch.action.admin.indices.stats.CommonStats; import org.opensearch.action.admin.indices.stats.CommonStatsFlags; import org.opensearch.action.admin.indices.stats.ShardStats; @@ -63,7 +64,10 @@ import java.io.IOException; import java.util.ArrayList; +import java.util.HashSet; import java.util.List; +import java.util.Map; +import java.util.Set; /** * Transport action for obtaining cluster state @@ -76,13 +80,19 @@ public class TransportClusterStatsAction extends TransportNodesAction< TransportClusterStatsAction.ClusterStatsNodeRequest, ClusterStatsNodeResponse> { - private static final CommonStatsFlags SHARD_STATS_FLAGS = new CommonStatsFlags( + private static final Map SHARDS_STATS_FLAG_MAP_TO_INDEX_METRIC = Map.of( CommonStatsFlags.Flag.Docs, + ClusterStatsRequest.IndexMetric.DOCS, CommonStatsFlags.Flag.Store, + ClusterStatsRequest.IndexMetric.STORE, CommonStatsFlags.Flag.FieldData, + ClusterStatsRequest.IndexMetric.FIELDDATA, CommonStatsFlags.Flag.QueryCache, + ClusterStatsRequest.IndexMetric.QUERY_CACHE, CommonStatsFlags.Flag.Completion, - CommonStatsFlags.Flag.Segments + ClusterStatsRequest.IndexMetric.COMPLETION, + CommonStatsFlags.Flag.Segments, + ClusterStatsRequest.IndexMetric.SEGMENTS ); private final NodeService nodeService; @@ -124,14 +134,27 @@ protected ClusterStatsResponse newResponse( + " the cluster state that are too slow for a transport thread" ); ClusterState state = clusterService.state(); - return new ClusterStatsResponse( - System.currentTimeMillis(), - state.metadata().clusterUUID(), - clusterService.getClusterName(), - responses, - failures, - state - ); + if (request.computeAllMetrics()) { + return new ClusterStatsResponse( + System.currentTimeMillis(), + state.metadata().clusterUUID(), + clusterService.getClusterName(), + responses, + failures, + state + ); + } else { + return new ClusterStatsResponse( + System.currentTimeMillis(), + state.metadata().clusterUUID(), + clusterService.getClusterName(), + responses, + failures, + state, + request.requestedMetrics(), + request.indicesMetrics() + ); + } } @Override @@ -149,17 +172,17 @@ protected ClusterStatsNodeResponse nodeOperation(ClusterStatsNodeRequest nodeReq NodeInfo nodeInfo = nodeService.info(true, true, false, true, false, true, false, true, false, false, false, false); NodeStats nodeStats = nodeService.stats( CommonStatsFlags.NONE, - true, - true, - true, + isMetricRequired(Metric.OS, nodeRequest.request), + isMetricRequired(Metric.PROCESS, nodeRequest.request), + isMetricRequired(Metric.JVM, nodeRequest.request), false, - true, + isMetricRequired(Metric.FS, nodeRequest.request), false, false, false, false, false, - true, + isMetricRequired(Metric.INGEST, nodeRequest.request), false, false, false, @@ -178,33 +201,36 @@ protected ClusterStatsNodeResponse nodeOperation(ClusterStatsNodeRequest nodeReq false ); List shardsStats = new ArrayList<>(); - for (IndexService indexService : indicesService) { - for (IndexShard indexShard : indexService) { - if (indexShard.routingEntry() != null && indexShard.routingEntry().active()) { - // only report on fully started shards - CommitStats commitStats; - SeqNoStats seqNoStats; - RetentionLeaseStats retentionLeaseStats; - try { - commitStats = indexShard.commitStats(); - seqNoStats = indexShard.seqNoStats(); - retentionLeaseStats = indexShard.getRetentionLeaseStats(); - } catch (final AlreadyClosedException e) { - // shard is closed - no stats is fine - commitStats = null; - seqNoStats = null; - retentionLeaseStats = null; + if (isMetricRequired(Metric.INDICES, nodeRequest.request)) { + CommonStatsFlags commonStatsFlags = getCommonStatsFlags(nodeRequest); + for (IndexService indexService : indicesService) { + for (IndexShard indexShard : indexService) { + if (indexShard.routingEntry() != null && indexShard.routingEntry().active()) { + // only report on fully started shards + CommitStats commitStats; + SeqNoStats seqNoStats; + RetentionLeaseStats retentionLeaseStats; + try { + commitStats = indexShard.commitStats(); + seqNoStats = indexShard.seqNoStats(); + retentionLeaseStats = indexShard.getRetentionLeaseStats(); + } catch (final AlreadyClosedException e) { + // shard is closed - no stats is fine + commitStats = null; + seqNoStats = null; + retentionLeaseStats = null; + } + shardsStats.add( + new ShardStats( + indexShard.routingEntry(), + indexShard.shardPath(), + new CommonStats(indicesService.getIndicesQueryCache(), indexShard, commonStatsFlags), + commitStats, + seqNoStats, + retentionLeaseStats + ) + ); } - shardsStats.add( - new ShardStats( - indexShard.routingEntry(), - indexShard.shardPath(), - new CommonStats(indicesService.getIndicesQueryCache(), indexShard, SHARD_STATS_FLAGS), - commitStats, - seqNoStats, - retentionLeaseStats - ) - ); } } } @@ -224,6 +250,31 @@ protected ClusterStatsNodeResponse nodeOperation(ClusterStatsNodeRequest nodeReq ); } + /** + * A metric is required when: all cluster stats are required (OR) if the metric is requested + * @param metric + * @param clusterStatsRequest + * @return + */ + private boolean isMetricRequired(Metric metric, ClusterStatsRequest clusterStatsRequest) { + return clusterStatsRequest.computeAllMetrics() || clusterStatsRequest.requestedMetrics().contains(metric); + } + + private static CommonStatsFlags getCommonStatsFlags(ClusterStatsNodeRequest nodeRequest) { + Set requestedCommonStatsFlags = new HashSet<>(); + if (nodeRequest.request.computeAllMetrics()) { + requestedCommonStatsFlags.addAll(SHARDS_STATS_FLAG_MAP_TO_INDEX_METRIC.keySet()); + } else { + for (Map.Entry entry : SHARDS_STATS_FLAG_MAP_TO_INDEX_METRIC + .entrySet()) { + if (nodeRequest.request.indicesMetrics().contains(entry.getValue())) { + requestedCommonStatsFlags.add(entry.getKey()); + } + } + } + return new CommonStatsFlags(requestedCommonStatsFlags.toArray(new CommonStatsFlags.Flag[0])); + } + /** * Inner Cluster Stats Node Request * diff --git a/server/src/main/java/org/opensearch/action/pagination/PaginationStrategy.java b/server/src/main/java/org/opensearch/action/pagination/PaginationStrategy.java index edb2489b1e4f0..8cc1ebdd2f194 100644 --- a/server/src/main/java/org/opensearch/action/pagination/PaginationStrategy.java +++ b/server/src/main/java/org/opensearch/action/pagination/PaginationStrategy.java @@ -44,8 +44,11 @@ public interface PaginationStrategy { List getRequestedEntities(); /** - * - * Utility method to get list of indices filtered as per {@param filterPredicate} and the sorted according to {@param comparator}. + * Utility method to get list of indices filtered and sorted as per the provided parameters. + * @param clusterState state consisting of all the indices to be filtered and sorted. + * @param filterPredicate predicate to be used for filtering out the required indices. + * @param comparator comparator to be used for sorting the already filtered list of indices. + * @return list of filtered and sorted IndexMetadata. */ static List getSortedIndexMetadata( final ClusterState clusterState, @@ -56,8 +59,10 @@ static List getSortedIndexMetadata( } /** - * - * Utility method to get list of indices sorted as per {@param comparator}. + * Utility method to get list of sorted indices. + * @param clusterState state consisting of indices to be sorted. + * @param comparator comparator to be used for sorting the list of indices. + * @return list of sorted IndexMetadata. */ static List getSortedIndexMetadata(final ClusterState clusterState, Comparator comparator) { return clusterState.metadata().indices().values().stream().sorted(comparator).collect(Collectors.toList()); diff --git a/server/src/main/java/org/opensearch/cluster/coordination/Coordinator.java b/server/src/main/java/org/opensearch/cluster/coordination/Coordinator.java index 684a6b0c3eae5..6fee2037501e7 100644 --- a/server/src/main/java/org/opensearch/cluster/coordination/Coordinator.java +++ b/server/src/main/java/org/opensearch/cluster/coordination/Coordinator.java @@ -141,7 +141,8 @@ public class Coordinator extends AbstractLifecycleComponent implements Discovery "cluster.publish.timeout", TimeValue.timeValueMillis(30000), TimeValue.timeValueMillis(1), - Setting.Property.NodeScope + Setting.Property.NodeScope, + Setting.Property.Dynamic ); private final Settings settings; @@ -164,7 +165,7 @@ public class Coordinator extends AbstractLifecycleComponent implements Discovery private final Random random; private final ElectionSchedulerFactory electionSchedulerFactory; private final SeedHostsResolver configuredHostsResolver; - private final TimeValue publishTimeout; + private TimeValue publishTimeout; private final TimeValue publishInfoTimeout; private final PublicationTransportHandler publicationHandler; private final LeaderChecker leaderChecker; @@ -247,6 +248,7 @@ public Coordinator( this.lastJoin = Optional.empty(); this.joinAccumulator = new InitialJoinAccumulator(); this.publishTimeout = PUBLISH_TIMEOUT_SETTING.get(settings); + clusterSettings.addSettingsUpdateConsumer(PUBLISH_TIMEOUT_SETTING, this::setPublishTimeout); this.publishInfoTimeout = PUBLISH_INFO_TIMEOUT_SETTING.get(settings); this.random = random; this.electionSchedulerFactory = new ElectionSchedulerFactory(settings, random, transportService.getThreadPool()); @@ -301,6 +303,7 @@ public Coordinator( ); this.lagDetector = new LagDetector( settings, + clusterSettings, transportService.getThreadPool(), n -> removeNode(n, "lagging"), transportService::getLocalNode @@ -319,6 +322,10 @@ public Coordinator( this.clusterSettings = clusterSettings; } + private void setPublishTimeout(TimeValue publishTimeout) { + this.publishTimeout = publishTimeout; + } + private ClusterFormationState getClusterFormationState() { return new ClusterFormationState( settings, @@ -1669,7 +1676,6 @@ public void onNodeAck(DiscoveryNode node, Exception e) { this.localNodeAckEvent = localNodeAckEvent; this.ackListener = ackListener; this.publishListener = publishListener; - this.timeoutHandler = singleNodeDiscovery ? null : transportService.getThreadPool().schedule(new Runnable() { @Override public void run() { diff --git a/server/src/main/java/org/opensearch/cluster/coordination/ElectionSchedulerFactory.java b/server/src/main/java/org/opensearch/cluster/coordination/ElectionSchedulerFactory.java index 828db5864d28b..1cc88c71c609b 100644 --- a/server/src/main/java/org/opensearch/cluster/coordination/ElectionSchedulerFactory.java +++ b/server/src/main/java/org/opensearch/cluster/coordination/ElectionSchedulerFactory.java @@ -214,7 +214,7 @@ protected void doRun() { if (isClosed.get()) { logger.debug("{} not starting election", this); } else { - logger.debug("{} starting election", this); + logger.debug("{} starting election with duration {}", this, duration); scheduleNextElection(duration, scheduledRunnable); scheduledRunnable.run(); } diff --git a/server/src/main/java/org/opensearch/cluster/coordination/FollowersChecker.java b/server/src/main/java/org/opensearch/cluster/coordination/FollowersChecker.java index 2ec0dabd91786..ca414ef7c4fc8 100644 --- a/server/src/main/java/org/opensearch/cluster/coordination/FollowersChecker.java +++ b/server/src/main/java/org/opensearch/cluster/coordination/FollowersChecker.java @@ -92,7 +92,8 @@ public class FollowersChecker { "cluster.fault_detection.follower_check.interval", TimeValue.timeValueMillis(1000), TimeValue.timeValueMillis(100), - Setting.Property.NodeScope + Setting.Property.NodeScope, + Setting.Property.Dynamic ); // the timeout for each check sent to each node @@ -100,7 +101,7 @@ public class FollowersChecker { "cluster.fault_detection.follower_check.timeout", TimeValue.timeValueMillis(10000), TimeValue.timeValueMillis(1), - TimeValue.timeValueMillis(60000), + TimeValue.timeValueMillis(150000), Setting.Property.NodeScope, Setting.Property.Dynamic ); @@ -115,7 +116,7 @@ public class FollowersChecker { private final Settings settings; - private final TimeValue followerCheckInterval; + private TimeValue followerCheckInterval; private TimeValue followerCheckTimeout; private final int followerCheckRetryCount; private final BiConsumer onNodeFailure; @@ -148,6 +149,7 @@ public FollowersChecker( followerCheckInterval = FOLLOWER_CHECK_INTERVAL_SETTING.get(settings); followerCheckTimeout = FOLLOWER_CHECK_TIMEOUT_SETTING.get(settings); followerCheckRetryCount = FOLLOWER_CHECK_RETRY_COUNT_SETTING.get(settings); + clusterSettings.addSettingsUpdateConsumer(FOLLOWER_CHECK_INTERVAL_SETTING, this::setFollowerCheckInterval); clusterSettings.addSettingsUpdateConsumer(FOLLOWER_CHECK_TIMEOUT_SETTING, this::setFollowerCheckTimeout); updateFastResponseState(0, Mode.CANDIDATE); transportService.registerRequestHandler( @@ -167,6 +169,10 @@ public void onNodeDisconnected(DiscoveryNode node, Transport.Connection connecti this.clusterManagerMetrics = clusterManagerMetrics; } + private void setFollowerCheckInterval(TimeValue followerCheckInterval) { + this.followerCheckInterval = followerCheckInterval; + } + private void setFollowerCheckTimeout(TimeValue followerCheckTimeout) { this.followerCheckTimeout = followerCheckTimeout; } diff --git a/server/src/main/java/org/opensearch/cluster/coordination/LagDetector.java b/server/src/main/java/org/opensearch/cluster/coordination/LagDetector.java index eeb0800663d0a..969c121dc87cf 100644 --- a/server/src/main/java/org/opensearch/cluster/coordination/LagDetector.java +++ b/server/src/main/java/org/opensearch/cluster/coordination/LagDetector.java @@ -34,6 +34,7 @@ import org.apache.logging.log4j.LogManager; import org.apache.logging.log4j.Logger; import org.opensearch.cluster.node.DiscoveryNode; +import org.opensearch.common.settings.ClusterSettings; import org.opensearch.common.settings.Setting; import org.opensearch.common.settings.Settings; import org.opensearch.common.unit.TimeValue; @@ -68,10 +69,11 @@ public class LagDetector { "cluster.follower_lag.timeout", TimeValue.timeValueMillis(90000), TimeValue.timeValueMillis(1), - Setting.Property.NodeScope + Setting.Property.NodeScope, + Setting.Property.Dynamic ); - private final TimeValue clusterStateApplicationTimeout; + private TimeValue clusterStateApplicationTimeout; private final Consumer onLagDetected; private final Supplier localNodeSupplier; private final ThreadPool threadPool; @@ -79,12 +81,14 @@ public class LagDetector { public LagDetector( final Settings settings, + final ClusterSettings clusterSettings, final ThreadPool threadPool, final Consumer onLagDetected, final Supplier localNodeSupplier ) { this.threadPool = threadPool; this.clusterStateApplicationTimeout = CLUSTER_FOLLOWER_LAG_TIMEOUT_SETTING.get(settings); + clusterSettings.addSettingsUpdateConsumer(CLUSTER_FOLLOWER_LAG_TIMEOUT_SETTING, this::setFollowerLagTimeout); this.onLagDetected = onLagDetected; this.localNodeSupplier = localNodeSupplier; } @@ -136,6 +140,10 @@ public String toString() { } } + private void setFollowerLagTimeout(TimeValue followerCheckLagTimeout) { + this.clusterStateApplicationTimeout = followerCheckLagTimeout; + } + @Override public String toString() { return "LagDetector{" diff --git a/server/src/main/java/org/opensearch/gateway/ShardsBatchGatewayAllocator.java b/server/src/main/java/org/opensearch/gateway/ShardsBatchGatewayAllocator.java index d7c0a66ba3424..9c38ea1df8a41 100644 --- a/server/src/main/java/org/opensearch/gateway/ShardsBatchGatewayAllocator.java +++ b/server/src/main/java/org/opensearch/gateway/ShardsBatchGatewayAllocator.java @@ -72,7 +72,7 @@ public class ShardsBatchGatewayAllocator implements ExistingShardsAllocator { public static final String ALLOCATOR_NAME = "shards_batch_gateway_allocator"; private static final Logger logger = LogManager.getLogger(ShardsBatchGatewayAllocator.class); - private final long maxBatchSize; + private long maxBatchSize; private static final short DEFAULT_SHARD_BATCH_SIZE = 2000; public static final String PRIMARY_BATCH_ALLOCATOR_TIMEOUT_SETTING_KEY = @@ -93,7 +93,8 @@ public class ShardsBatchGatewayAllocator implements ExistingShardsAllocator { DEFAULT_SHARD_BATCH_SIZE, 1, 10000, - Setting.Property.NodeScope + Setting.Property.NodeScope, + Setting.Property.Dynamic ); /** @@ -172,6 +173,7 @@ public ShardsBatchGatewayAllocator( this.batchStartedAction = batchStartedAction; this.batchStoreAction = batchStoreAction; this.maxBatchSize = GATEWAY_ALLOCATOR_BATCH_SIZE.get(settings); + clusterSettings.addSettingsUpdateConsumer(GATEWAY_ALLOCATOR_BATCH_SIZE, this::setMaxBatchSize); this.primaryShardsBatchGatewayAllocatorTimeout = PRIMARY_BATCH_ALLOCATOR_TIMEOUT_SETTING.get(settings); clusterSettings.addSettingsUpdateConsumer(PRIMARY_BATCH_ALLOCATOR_TIMEOUT_SETTING, this::setPrimaryBatchAllocatorTimeout); this.replicaShardsBatchGatewayAllocatorTimeout = REPLICA_BATCH_ALLOCATOR_TIMEOUT_SETTING.get(settings); @@ -402,6 +404,7 @@ else if (shardRouting.primary() == primary) { Iterator iterator = newShardsToBatch.values().iterator(); assert maxBatchSize > 0 : "Shards batch size must be greater than 0"; + logger.debug("Using async fetch batch size {}", maxBatchSize); long batchSize = maxBatchSize; Map perBatchShards = new HashMap<>(); while (iterator.hasNext()) { @@ -906,6 +909,10 @@ public int getNumberOfStoreShardBatches() { return batchIdToStoreShardBatch.size(); } + private void setMaxBatchSize(long maxBatchSize) { + this.maxBatchSize = maxBatchSize; + } + protected void setPrimaryBatchAllocatorTimeout(TimeValue primaryShardsBatchGatewayAllocatorTimeout) { this.primaryShardsBatchGatewayAllocatorTimeout = primaryShardsBatchGatewayAllocatorTimeout; } diff --git a/server/src/main/java/org/opensearch/index/cache/query/QueryCacheStats.java b/server/src/main/java/org/opensearch/index/cache/query/QueryCacheStats.java index d844e5cbb8897..8ca3157d9f775 100644 --- a/server/src/main/java/org/opensearch/index/cache/query/QueryCacheStats.java +++ b/server/src/main/java/org/opensearch/index/cache/query/QueryCacheStats.java @@ -77,6 +77,9 @@ public QueryCacheStats(long ramBytesUsed, long hitCount, long missCount, long ca } public void add(QueryCacheStats stats) { + if (stats == null) { + return; + } ramBytesUsed += stats.ramBytesUsed; hitCount += stats.hitCount; missCount += stats.missCount; diff --git a/server/src/main/java/org/opensearch/index/fielddata/FieldDataStats.java b/server/src/main/java/org/opensearch/index/fielddata/FieldDataStats.java index 85b435e969bfa..1bd81c840a3d4 100644 --- a/server/src/main/java/org/opensearch/index/fielddata/FieldDataStats.java +++ b/server/src/main/java/org/opensearch/index/fielddata/FieldDataStats.java @@ -80,6 +80,9 @@ public FieldDataStats(long memorySize, long evictions, @Nullable FieldMemoryStat } public void add(FieldDataStats stats) { + if (stats == null) { + return; + } this.memorySize += stats.memorySize; this.evictions += stats.evictions; if (stats.fields != null) { diff --git a/server/src/main/java/org/opensearch/indices/replication/PrimaryShardReplicationSource.java b/server/src/main/java/org/opensearch/indices/replication/PrimaryShardReplicationSource.java index a17779810239a..af37594f88fee 100644 --- a/server/src/main/java/org/opensearch/indices/replication/PrimaryShardReplicationSource.java +++ b/server/src/main/java/org/opensearch/indices/replication/PrimaryShardReplicationSource.java @@ -47,6 +47,8 @@ public PrimaryShardReplicationSource( RecoverySettings recoverySettings, DiscoveryNode sourceNode ) { + assert targetNode != null : "Target node must be set"; + assert sourceNode != null : "Source node must be set"; this.targetAllocationId = targetAllocationId; this.transportService = transportService; this.sourceNode = sourceNode; diff --git a/server/src/main/java/org/opensearch/indices/replication/SegmentReplicationSourceFactory.java b/server/src/main/java/org/opensearch/indices/replication/SegmentReplicationSourceFactory.java index 81eb38757aebe..2b6723512abf2 100644 --- a/server/src/main/java/org/opensearch/indices/replication/SegmentReplicationSourceFactory.java +++ b/server/src/main/java/org/opensearch/indices/replication/SegmentReplicationSourceFactory.java @@ -53,6 +53,10 @@ public SegmentReplicationSource get(IndexShard shard) { private DiscoveryNode getPrimaryNode(ShardId shardId) { ShardRouting primaryShard = clusterService.state().routingTable().shardRoutingTable(shardId).primaryShard(); - return clusterService.state().nodes().get(primaryShard.currentNodeId()); + DiscoveryNode node = clusterService.state().nodes().get(primaryShard.currentNodeId()); + if (node == null) { + throw new IllegalStateException("Cannot replicate, primary shard for " + shardId + " is not allocated on any node"); + } + return node; } } diff --git a/server/src/main/java/org/opensearch/rest/action/admin/cluster/RestClusterStatsAction.java b/server/src/main/java/org/opensearch/rest/action/admin/cluster/RestClusterStatsAction.java index ee33bd18db05d..47f3e048c516a 100644 --- a/server/src/main/java/org/opensearch/rest/action/admin/cluster/RestClusterStatsAction.java +++ b/server/src/main/java/org/opensearch/rest/action/admin/cluster/RestClusterStatsAction.java @@ -33,13 +33,24 @@ package org.opensearch.rest.action.admin.cluster; import org.opensearch.action.admin.cluster.stats.ClusterStatsRequest; +import org.opensearch.action.admin.cluster.stats.ClusterStatsRequest.IndexMetric; +import org.opensearch.action.admin.cluster.stats.ClusterStatsRequest.Metric; import org.opensearch.client.node.NodeClient; +import org.opensearch.core.common.Strings; import org.opensearch.rest.BaseRestHandler; import org.opensearch.rest.RestRequest; import org.opensearch.rest.action.RestActions.NodesResponseRestListener; import java.io.IOException; +import java.util.Collections; +import java.util.HashMap; +import java.util.HashSet; import java.util.List; +import java.util.Locale; +import java.util.Map; +import java.util.Set; +import java.util.TreeSet; +import java.util.function.Consumer; import static java.util.Arrays.asList; import static java.util.Collections.unmodifiableList; @@ -54,7 +65,34 @@ public class RestClusterStatsAction extends BaseRestHandler { @Override public List routes() { - return unmodifiableList(asList(new Route(GET, "/_cluster/stats"), new Route(GET, "/_cluster/stats/nodes/{nodeId}"))); + return unmodifiableList( + asList( + new Route(GET, "/_cluster/stats"), + new Route(GET, "/_cluster/stats/nodes/{nodeId}"), + new Route(GET, "/_cluster/stats/{metric}/nodes/{nodeId}"), + new Route(GET, "/_cluster/stats/{metric}/{index_metric}/nodes/{nodeId}") + ) + ); + } + + static final Map> INDEX_METRIC_TO_REQUEST_CONSUMER_MAP; + + static final Map> METRIC_REQUEST_CONSUMER_MAP; + + static { + Map> metricRequestConsumerMap = new HashMap<>(); + for (Metric metric : Metric.values()) { + metricRequestConsumerMap.put(metric.metricName(), request -> request.addMetric(metric)); + } + METRIC_REQUEST_CONSUMER_MAP = Collections.unmodifiableMap(metricRequestConsumerMap); + } + + static { + Map> metricMap = new HashMap<>(); + for (IndexMetric indexMetric : IndexMetric.values()) { + metricMap.put(indexMetric.metricName(), request -> request.addIndexMetric(indexMetric)); + } + INDEX_METRIC_TO_REQUEST_CONSUMER_MAP = Collections.unmodifiableMap(metricMap); } @Override @@ -64,10 +102,101 @@ public String getName() { @Override public RestChannelConsumer prepareRequest(final RestRequest request, final NodeClient client) throws IOException { - ClusterStatsRequest clusterStatsRequest = new ClusterStatsRequest().nodesIds(request.paramAsStringArray("nodeId", null)); + ClusterStatsRequest clusterStatsRequest = fromRequest(request); + return channel -> client.admin().cluster().clusterStats(clusterStatsRequest, new NodesResponseRestListener<>(channel)); + } + + public static ClusterStatsRequest fromRequest(final RestRequest request) { + Set metrics = Strings.tokenizeByCommaToSet(request.param("metric", "_all")); + // Value for param index_metric defaults to _all when indices metric or all metrics are requested. + String indicesMetricsDefaultValue = metrics.contains(Metric.INDICES.metricName()) || metrics.contains("_all") ? "_all" : null; + Set indexMetrics = Strings.tokenizeByCommaToSet(request.param("index_metric", indicesMetricsDefaultValue)); + String[] nodeIds = request.paramAsStringArray("nodeId", null); + + ClusterStatsRequest clusterStatsRequest = new ClusterStatsRequest().nodesIds(nodeIds); clusterStatsRequest.timeout(request.param("timeout")); clusterStatsRequest.useAggregatedNodeLevelResponses(true); - return channel -> client.admin().cluster().clusterStats(clusterStatsRequest, new NodesResponseRestListener<>(channel)); + clusterStatsRequest.computeAllMetrics(false); + + paramValidations(metrics, indexMetrics, request); + final Set metricsRequested = metrics.contains("_all") + ? new HashSet<>(METRIC_REQUEST_CONSUMER_MAP.keySet()) + : new HashSet<>(metrics); + Set invalidMetrics = validateAndSetRequestedMetrics(metricsRequested, METRIC_REQUEST_CONSUMER_MAP, clusterStatsRequest); + if (!invalidMetrics.isEmpty()) { + throw new IllegalArgumentException( + unrecognizedStrings(request, invalidMetrics, METRIC_REQUEST_CONSUMER_MAP.keySet(), "metric") + ); + } + if (metricsRequested.contains(Metric.INDICES.metricName())) { + final Set indexMetricsRequested = indexMetrics.contains("_all") + ? INDEX_METRIC_TO_REQUEST_CONSUMER_MAP.keySet() + : new HashSet<>(indexMetrics); + Set invalidIndexMetrics = validateAndSetRequestedMetrics( + indexMetricsRequested, + INDEX_METRIC_TO_REQUEST_CONSUMER_MAP, + clusterStatsRequest + ); + if (!invalidIndexMetrics.isEmpty()) { + throw new IllegalArgumentException( + unrecognizedStrings(request, invalidIndexMetrics, INDEX_METRIC_TO_REQUEST_CONSUMER_MAP.keySet(), "index metric") + ); + } + } + + return clusterStatsRequest; + } + + private static void paramValidations(Set metrics, Set indexMetrics, RestRequest request) { + if (metrics.size() > 1 && metrics.contains("_all")) { + throw new IllegalArgumentException( + String.format( + Locale.ROOT, + "request [%s] contains _all and individual metrics [%s]", + request.path(), + request.param("metric") + ) + ); + } + + if (indexMetrics.size() > 1 && indexMetrics.contains("_all")) { + throw new IllegalArgumentException( + String.format( + Locale.ROOT, + "request [%s] contains _all and individual index metrics [%s]", + request.path(), + request.param("index_metric") + ) + ); + } + + if (!metrics.contains(Metric.INDICES.metricName()) && !metrics.contains("_all") && !indexMetrics.isEmpty()) { + throw new IllegalArgumentException( + String.format( + Locale.ROOT, + "request [%s] contains index metrics [%s] but indices stats not requested", + request.path(), + request.param("index_metric") + ) + ); + } + } + + private static Set validateAndSetRequestedMetrics( + Set metrics, + Map> metricConsumerMap, + ClusterStatsRequest clusterStatsRequest + ) { + final Set invalidMetrics = new TreeSet<>(); + for (String metric : metrics) { + Consumer clusterStatsRequestConsumer = metricConsumerMap.get(metric); + if (clusterStatsRequestConsumer != null) { + clusterStatsRequestConsumer.accept(clusterStatsRequest); + } else { + invalidMetrics.add(metric); + } + } + return invalidMetrics; } @Override diff --git a/server/src/main/java/org/opensearch/wlm/QueryGroupService.java b/server/src/main/java/org/opensearch/wlm/QueryGroupService.java index cb0be5597766a..14002a2b38134 100644 --- a/server/src/main/java/org/opensearch/wlm/QueryGroupService.java +++ b/server/src/main/java/org/opensearch/wlm/QueryGroupService.java @@ -266,11 +266,12 @@ public void rejectIfNeeded(String queryGroupId) { return; } - // rejections will not happen for SOFT mode QueryGroups + // rejections will not happen for SOFT mode QueryGroups unless node is in duress Optional optionalQueryGroup = activeQueryGroups.stream().filter(x -> x.get_id().equals(queryGroupId)).findFirst(); - if (optionalQueryGroup.isPresent() && optionalQueryGroup.get().getResiliencyMode() == MutableQueryGroupFragment.ResiliencyMode.SOFT) - return; + if (optionalQueryGroup.isPresent() + && (optionalQueryGroup.get().getResiliencyMode() == MutableQueryGroupFragment.ResiliencyMode.SOFT + && !nodeDuressTrackers.isNodeInDuress())) return; optionalQueryGroup.ifPresent(queryGroup -> { boolean reject = false; diff --git a/server/src/test/java/org/opensearch/action/admin/cluster/stats/ClusterStatsResponseTests.java b/server/src/test/java/org/opensearch/action/admin/cluster/stats/ClusterStatsResponseTests.java new file mode 100644 index 0000000000000..ad7706292d93c --- /dev/null +++ b/server/src/test/java/org/opensearch/action/admin/cluster/stats/ClusterStatsResponseTests.java @@ -0,0 +1,281 @@ +/* + * SPDX-License-Identifier: Apache-2.0 + * + * The OpenSearch Contributors require contributions made to + * this file be licensed under the Apache-2.0 license or a + * compatible open source license. + */ + +package org.opensearch.action.admin.cluster.stats; + +import org.opensearch.Build; +import org.opensearch.Version; +import org.opensearch.action.admin.cluster.node.info.NodeInfo; +import org.opensearch.action.admin.cluster.node.info.PluginsAndModules; +import org.opensearch.action.admin.cluster.node.stats.NodeStats; +import org.opensearch.action.admin.cluster.stats.ClusterStatsRequest.IndexMetric; +import org.opensearch.action.admin.indices.stats.CommonStats; +import org.opensearch.action.admin.indices.stats.CommonStatsFlags; +import org.opensearch.action.admin.indices.stats.ShardStats; +import org.opensearch.cluster.ClusterName; +import org.opensearch.cluster.ClusterState; +import org.opensearch.cluster.node.DiscoveryNode; +import org.opensearch.cluster.routing.ShardRouting; +import org.opensearch.cluster.routing.ShardRoutingState; +import org.opensearch.cluster.routing.TestShardRouting; +import org.opensearch.common.io.stream.BytesStreamOutput; +import org.opensearch.common.settings.Settings; +import org.opensearch.core.common.io.stream.StreamInput; +import org.opensearch.core.common.transport.BoundTransportAddress; +import org.opensearch.core.common.transport.TransportAddress; +import org.opensearch.core.index.Index; +import org.opensearch.index.cache.query.QueryCacheStats; +import org.opensearch.index.engine.SegmentsStats; +import org.opensearch.index.fielddata.FieldDataStats; +import org.opensearch.index.flush.FlushStats; +import org.opensearch.index.shard.DocsStats; +import org.opensearch.index.shard.IndexingStats; +import org.opensearch.index.shard.ShardPath; +import org.opensearch.index.store.StoreStats; +import org.opensearch.monitor.jvm.JvmInfo; +import org.opensearch.monitor.jvm.JvmStats; +import org.opensearch.monitor.os.OsInfo; +import org.opensearch.monitor.process.ProcessStats; +import org.opensearch.search.suggest.completion.CompletionStats; +import org.opensearch.test.OpenSearchTestCase; +import org.opensearch.transport.TransportInfo; + +import java.io.IOException; +import java.nio.file.Path; +import java.util.ArrayList; +import java.util.Collections; +import java.util.List; +import java.util.Set; + +public class ClusterStatsResponseTests extends OpenSearchTestCase { + + public void testSerializationWithIndicesMappingAndAnalysisStats() throws Exception { + List defaultClusterStatsNodeResponses = new ArrayList<>(); + + int numberOfNodes = randomIntBetween(1, 4); + Index testIndex = new Index("test-index", "_na_"); + + for (int i = 0; i < numberOfNodes; i++) { + DiscoveryNode node = new DiscoveryNode("node-" + i, buildNewFakeTransportAddress(), Version.CURRENT); + CommonStats commonStats = createRandomCommonStats(); + ShardStats[] shardStats = createShardStats(node, testIndex, commonStats); + ClusterStatsNodeResponse customClusterStatsResponse = createClusterStatsNodeResponse(node, shardStats); + defaultClusterStatsNodeResponses.add(customClusterStatsResponse); + } + ClusterStatsResponse clusterStatsResponse = new ClusterStatsResponse( + 1l, + "UUID", + new ClusterName("cluster_name"), + defaultClusterStatsNodeResponses, + List.of(), + ClusterState.EMPTY_STATE, + Set.of(ClusterStatsRequest.Metric.INDICES), + Set.of(IndexMetric.MAPPINGS, IndexMetric.ANALYSIS) + ); + BytesStreamOutput output = new BytesStreamOutput(); + clusterStatsResponse.writeTo(output); + + StreamInput streamInput = output.bytes().streamInput(); + ClusterStatsResponse deserializedClusterStatsResponse = new ClusterStatsResponse(streamInput); + assertEquals(clusterStatsResponse.timestamp, deserializedClusterStatsResponse.timestamp); + assertEquals(clusterStatsResponse.status, deserializedClusterStatsResponse.status); + assertEquals(clusterStatsResponse.clusterUUID, deserializedClusterStatsResponse.clusterUUID); + assertNotNull(clusterStatsResponse.indicesStats); + assertEquals(clusterStatsResponse.indicesStats.getMappings(), deserializedClusterStatsResponse.indicesStats.getMappings()); + assertEquals(clusterStatsResponse.indicesStats.getAnalysis(), deserializedClusterStatsResponse.indicesStats.getAnalysis()); + } + + public void testSerializationWithoutIndicesMappingAndAnalysisStats() throws Exception { + List defaultClusterStatsNodeResponses = new ArrayList<>(); + + int numberOfNodes = randomIntBetween(1, 4); + Index testIndex = new Index("test-index", "_na_"); + + for (int i = 0; i < numberOfNodes; i++) { + DiscoveryNode node = new DiscoveryNode("node-" + i, buildNewFakeTransportAddress(), Version.CURRENT); + CommonStats commonStats = createRandomCommonStats(); + ShardStats[] shardStats = createShardStats(node, testIndex, commonStats); + ClusterStatsNodeResponse customClusterStatsResponse = createClusterStatsNodeResponse(node, shardStats); + defaultClusterStatsNodeResponses.add(customClusterStatsResponse); + } + ClusterStatsResponse clusterStatsResponse = new ClusterStatsResponse( + 1l, + "UUID", + new ClusterName("cluster_name"), + defaultClusterStatsNodeResponses, + List.of(), + ClusterState.EMPTY_STATE, + Set.of(ClusterStatsRequest.Metric.INDICES, ClusterStatsRequest.Metric.PROCESS, ClusterStatsRequest.Metric.JVM), + Set.of( + IndexMetric.DOCS, + IndexMetric.STORE, + IndexMetric.SEGMENTS, + IndexMetric.QUERY_CACHE, + IndexMetric.FIELDDATA, + IndexMetric.COMPLETION + ) + ); + BytesStreamOutput output = new BytesStreamOutput(); + clusterStatsResponse.writeTo(output); + + StreamInput streamInput = output.bytes().streamInput(); + ClusterStatsResponse deserializedClusterStatsResponse = new ClusterStatsResponse(streamInput); + assertEquals(clusterStatsResponse.timestamp, deserializedClusterStatsResponse.timestamp); + assertEquals(clusterStatsResponse.status, deserializedClusterStatsResponse.status); + assertEquals(clusterStatsResponse.clusterUUID, deserializedClusterStatsResponse.clusterUUID); + assertNotNull(deserializedClusterStatsResponse.nodesStats); + assertNotNull(deserializedClusterStatsResponse.nodesStats.getProcess()); + assertNotNull(deserializedClusterStatsResponse.nodesStats.getJvm()); + assertNotNull(deserializedClusterStatsResponse.indicesStats); + assertNotNull(deserializedClusterStatsResponse.indicesStats.getDocs()); + assertNotNull(deserializedClusterStatsResponse.indicesStats.getStore()); + assertNotNull(deserializedClusterStatsResponse.indicesStats.getSegments()); + assertNotNull(deserializedClusterStatsResponse.indicesStats.getQueryCache()); + assertNotNull(deserializedClusterStatsResponse.indicesStats.getFieldData()); + assertNotNull(deserializedClusterStatsResponse.indicesStats.getCompletion()); + assertNull(deserializedClusterStatsResponse.indicesStats.getMappings()); + assertNull(deserializedClusterStatsResponse.indicesStats.getAnalysis()); + } + + private ClusterStatsNodeResponse createClusterStatsNodeResponse(DiscoveryNode node, ShardStats[] shardStats) throws IOException { + JvmStats.GarbageCollector[] garbageCollectorsArray = new JvmStats.GarbageCollector[1]; + garbageCollectorsArray[0] = new JvmStats.GarbageCollector( + randomAlphaOfLengthBetween(3, 10), + randomNonNegativeLong(), + randomNonNegativeLong() + ); + JvmStats.GarbageCollectors garbageCollectors = new JvmStats.GarbageCollectors(garbageCollectorsArray); + NodeInfo nodeInfo = new NodeInfo( + Version.CURRENT, + Build.CURRENT, + node, + Settings.EMPTY, + new OsInfo(randomLong(), randomInt(), randomInt(), "name", "pretty_name", "arch", "version"), + null, + JvmInfo.jvmInfo(), + null, + new TransportInfo( + new BoundTransportAddress(new TransportAddress[] { buildNewFakeTransportAddress() }, buildNewFakeTransportAddress()), + null + ), + null, + new PluginsAndModules(Collections.emptyList(), Collections.emptyList()), + null, + null, + null, + null + ); + + NodeStats nodeStats = new NodeStats( + node, + randomNonNegativeLong(), + null, + null, + new ProcessStats( + randomNonNegativeLong(), + randomNonNegativeLong(), + randomNonNegativeLong(), + new ProcessStats.Cpu(randomShort(), randomNonNegativeLong()), + new ProcessStats.Mem(randomNonNegativeLong()) + ), + new JvmStats( + randomNonNegativeLong(), + randomNonNegativeLong(), + new JvmStats.Mem( + randomNonNegativeLong(), + randomNonNegativeLong(), + randomNonNegativeLong(), + randomNonNegativeLong(), + randomNonNegativeLong(), + Collections.emptyList() + ), + new JvmStats.Threads(randomIntBetween(1, 1000), randomIntBetween(1, 1000)), + garbageCollectors, + Collections.emptyList(), + new JvmStats.Classes(randomNonNegativeLong(), randomNonNegativeLong(), randomNonNegativeLong()) + ), + null, + null, + null, + null, + null, + null, + null, + null, + null, + null, + null, + null, + null, + null, + null, + null, + null, + null, + null, + null, + null, + null, + null, + null + ); + return new ClusterStatsNodeResponse(node, null, nodeInfo, nodeStats, shardStats); + + } + + private CommonStats createRandomCommonStats() { + CommonStats commonStats = new CommonStats(CommonStatsFlags.NONE); + commonStats.docs = new DocsStats(randomLongBetween(0, 10000), randomLongBetween(0, 100), randomLongBetween(0, 1000)); + commonStats.store = new StoreStats(randomLongBetween(0, 100), randomLongBetween(0, 1000)); + commonStats.indexing = new IndexingStats(); + commonStats.completion = new CompletionStats(); + commonStats.flush = new FlushStats(randomLongBetween(0, 100), randomLongBetween(0, 100), randomLongBetween(0, 100)); + commonStats.fieldData = new FieldDataStats(randomLongBetween(0, 100), randomLongBetween(0, 100), null); + commonStats.queryCache = new QueryCacheStats( + randomLongBetween(0, 100), + randomLongBetween(0, 100), + randomLongBetween(0, 100), + randomLongBetween(0, 100), + randomLongBetween(0, 100) + ); + commonStats.segments = new SegmentsStats(); + + return commonStats; + } + + private ShardStats[] createShardStats(DiscoveryNode localNode, Index index, CommonStats commonStats) { + List shardStatsList = new ArrayList<>(); + for (int i = 0; i < 2; i++) { + ShardRoutingState shardRoutingState = ShardRoutingState.fromValue((byte) randomIntBetween(2, 3)); + ShardRouting shardRouting = TestShardRouting.newShardRouting( + index.getName(), + i, + localNode.getId(), + randomBoolean(), + shardRoutingState + ); + + Path path = createTempDir().resolve("indices") + .resolve(shardRouting.shardId().getIndex().getUUID()) + .resolve(String.valueOf(shardRouting.shardId().id())); + + ShardStats shardStats = new ShardStats( + shardRouting, + new ShardPath(false, path, path, shardRouting.shardId()), + commonStats, + null, + null, + null + ); + shardStatsList.add(shardStats); + } + + return shardStatsList.toArray(new ShardStats[0]); + } + +} diff --git a/server/src/test/java/org/opensearch/cluster/coordination/CoordinationCheckerSettingsTests.java b/server/src/test/java/org/opensearch/cluster/coordination/CoordinationCheckerSettingsTests.java index 56bd2d94dce84..8e8e71ad33e75 100644 --- a/server/src/test/java/org/opensearch/cluster/coordination/CoordinationCheckerSettingsTests.java +++ b/server/src/test/java/org/opensearch/cluster/coordination/CoordinationCheckerSettingsTests.java @@ -14,7 +14,10 @@ import org.opensearch.common.unit.TimeValue; import org.opensearch.test.OpenSearchSingleNodeTestCase; +import static org.opensearch.cluster.coordination.Coordinator.PUBLISH_TIMEOUT_SETTING; +import static org.opensearch.cluster.coordination.FollowersChecker.FOLLOWER_CHECK_INTERVAL_SETTING; import static org.opensearch.cluster.coordination.FollowersChecker.FOLLOWER_CHECK_TIMEOUT_SETTING; +import static org.opensearch.cluster.coordination.LagDetector.CLUSTER_FOLLOWER_LAG_TIMEOUT_SETTING; import static org.opensearch.cluster.coordination.LeaderChecker.LEADER_CHECK_TIMEOUT_SETTING; import static org.opensearch.common.unit.TimeValue.timeValueSeconds; import static org.opensearch.test.hamcrest.OpenSearchAssertions.assertAcked; @@ -42,10 +45,10 @@ public void testFollowerCheckTimeoutValueUpdate() { public void testFollowerCheckTimeoutMaxValue() { Setting setting1 = FOLLOWER_CHECK_TIMEOUT_SETTING; - Settings timeSettings1 = Settings.builder().put(setting1.getKey(), "61s").build(); + Settings timeSettings1 = Settings.builder().put(setting1.getKey(), "151s").build(); assertThrows( - "failed to parse value [61s] for setting [" + setting1.getKey() + "], must be <= [60000ms]", + "failed to parse value [151s] for setting [" + setting1.getKey() + "], must be <= [150000ms]", IllegalArgumentException.class, () -> { client().admin().cluster().prepareUpdateSettings().setPersistentSettings(timeSettings1).execute().actionGet(); @@ -66,6 +69,38 @@ public void testFollowerCheckTimeoutMinValue() { ); } + public void testFollowerCheckIntervalValueUpdate() { + Setting setting1 = FOLLOWER_CHECK_INTERVAL_SETTING; + Settings timeSettings1 = Settings.builder().put(setting1.getKey(), "10s").build(); + try { + ClusterUpdateSettingsResponse response = client().admin() + .cluster() + .prepareUpdateSettings() + .setPersistentSettings(timeSettings1) + .execute() + .actionGet(); + assertAcked(response); + assertEquals(timeValueSeconds(10), setting1.get(response.getPersistentSettings())); + } finally { + // cleanup + timeSettings1 = Settings.builder().putNull(setting1.getKey()).build(); + client().admin().cluster().prepareUpdateSettings().setPersistentSettings(timeSettings1).execute().actionGet(); + } + } + + public void testFollowerCheckIntervalMinValue() { + Setting setting1 = FOLLOWER_CHECK_INTERVAL_SETTING; + Settings timeSettings1 = Settings.builder().put(setting1.getKey(), "10ms").build(); + + assertThrows( + "failed to parse value [10ms] for setting [" + setting1.getKey() + "], must be >= [100ms]", + IllegalArgumentException.class, + () -> { + client().admin().cluster().prepareUpdateSettings().setPersistentSettings(timeSettings1).execute().actionGet(); + } + ); + } + public void testLeaderCheckTimeoutValueUpdate() { Setting setting1 = LEADER_CHECK_TIMEOUT_SETTING; Settings timeSettings1 = Settings.builder().put(setting1.getKey(), "60s").build(); @@ -110,4 +145,70 @@ public void testLeaderCheckTimeoutMinValue() { } ); } + + public void testClusterPublishTimeoutValueUpdate() { + Setting setting1 = PUBLISH_TIMEOUT_SETTING; + Settings timeSettings1 = Settings.builder().put(setting1.getKey(), "60s").build(); + try { + ClusterUpdateSettingsResponse response = client().admin() + .cluster() + .prepareUpdateSettings() + .setPersistentSettings(timeSettings1) + .execute() + .actionGet(); + assertAcked(response); + assertEquals(timeValueSeconds(60), setting1.get(response.getPersistentSettings())); + } finally { + // cleanup + timeSettings1 = Settings.builder().putNull(setting1.getKey()).build(); + client().admin().cluster().prepareUpdateSettings().setPersistentSettings(timeSettings1).execute().actionGet(); + } + } + + public void testClusterPublishTimeoutMinValue() { + Setting setting1 = PUBLISH_TIMEOUT_SETTING; + Settings timeSettings1 = Settings.builder().put(setting1.getKey(), "0s").build(); + + assertThrows( + "failed to parse value [0s] for setting [" + setting1.getKey() + "], must be >= [1ms]", + IllegalArgumentException.class, + () -> { + client().admin().cluster().prepareUpdateSettings().setPersistentSettings(timeSettings1).execute().actionGet(); + } + ); + } + + public void testLagDetectorTimeoutUpdate() { + Setting setting1 = CLUSTER_FOLLOWER_LAG_TIMEOUT_SETTING; + Settings lagDetectorTimeout = Settings.builder().put(setting1.getKey(), "30s").build(); + try { + ClusterUpdateSettingsResponse response = client().admin() + .cluster() + .prepareUpdateSettings() + .setPersistentSettings(lagDetectorTimeout) + .execute() + .actionGet(); + + assertAcked(response); + assertEquals(timeValueSeconds(30), setting1.get(response.getPersistentSettings())); + } finally { + // cleanup + lagDetectorTimeout = Settings.builder().putNull(setting1.getKey()).build(); + client().admin().cluster().prepareUpdateSettings().setPersistentSettings(lagDetectorTimeout).execute().actionGet(); + } + } + + public void testLagDetectorTimeoutMinValue() { + Setting setting1 = CLUSTER_FOLLOWER_LAG_TIMEOUT_SETTING; + Settings lagDetectorTimeout = Settings.builder().put(setting1.getKey(), "0s").build(); + + assertThrows( + "failed to parse value [0s] for setting [" + setting1.getKey() + "], must be >= [1ms]", + IllegalArgumentException.class, + () -> { + client().admin().cluster().prepareUpdateSettings().setPersistentSettings(lagDetectorTimeout).execute().actionGet(); + } + ); + } + } diff --git a/server/src/test/java/org/opensearch/cluster/coordination/LagDetectorTests.java b/server/src/test/java/org/opensearch/cluster/coordination/LagDetectorTests.java index adffa27e9bc1a..315e5d6224227 100644 --- a/server/src/test/java/org/opensearch/cluster/coordination/LagDetectorTests.java +++ b/server/src/test/java/org/opensearch/cluster/coordination/LagDetectorTests.java @@ -32,6 +32,7 @@ package org.opensearch.cluster.coordination; import org.opensearch.cluster.node.DiscoveryNode; +import org.opensearch.common.settings.ClusterSettings; import org.opensearch.common.settings.Settings; import org.opensearch.common.unit.TimeValue; import org.opensearch.test.OpenSearchTestCase; @@ -70,8 +71,9 @@ public void setupFixture() { } else { followerLagTimeout = CLUSTER_FOLLOWER_LAG_TIMEOUT_SETTING.get(Settings.EMPTY); } - - lagDetector = new LagDetector(settingsBuilder.build(), deterministicTaskQueue.getThreadPool(), failedNodes::add, () -> localNode); + Settings settings = settingsBuilder.build(); + final ClusterSettings clusterSettings = new ClusterSettings(settings, ClusterSettings.BUILT_IN_CLUSTER_SETTINGS); + lagDetector = new LagDetector(settings, clusterSettings, deterministicTaskQueue.getThreadPool(), failedNodes::add, () -> localNode); localNode = CoordinationStateTests.createNode("local"); node1 = CoordinationStateTests.createNode("node1"); diff --git a/server/src/test/java/org/opensearch/gateway/ShardsBatchGatewayAllocatorTests.java b/server/src/test/java/org/opensearch/gateway/ShardsBatchGatewayAllocatorTests.java new file mode 100644 index 0000000000000..59fb6e2b940ba --- /dev/null +++ b/server/src/test/java/org/opensearch/gateway/ShardsBatchGatewayAllocatorTests.java @@ -0,0 +1,66 @@ +/* + * SPDX-License-Identifier: Apache-2.0 + * + * The OpenSearch Contributors require contributions made to + * this file be licensed under the Apache-2.0 license or a + * compatible open source license. + */ + +package org.opensearch.gateway; + +import org.opensearch.action.admin.cluster.settings.ClusterUpdateSettingsResponse; +import org.opensearch.common.settings.Setting; +import org.opensearch.common.settings.Settings; +import org.opensearch.test.OpenSearchSingleNodeTestCase; + +import static org.opensearch.gateway.ShardsBatchGatewayAllocator.GATEWAY_ALLOCATOR_BATCH_SIZE; +import static org.opensearch.test.hamcrest.OpenSearchAssertions.assertAcked; +import static org.hamcrest.Matchers.equalTo; + +public class ShardsBatchGatewayAllocatorTests extends OpenSearchSingleNodeTestCase { + public void testBatchSizeValueUpdate() { + Setting setting1 = GATEWAY_ALLOCATOR_BATCH_SIZE; + Settings batchSizeSetting = Settings.builder().put(setting1.getKey(), "3000").build(); + try { + ClusterUpdateSettingsResponse response = client().admin() + .cluster() + .prepareUpdateSettings() + .setPersistentSettings(batchSizeSetting) + .execute() + .actionGet(); + + assertAcked(response); + assertThat(setting1.get(response.getPersistentSettings()), equalTo(3000L)); + } finally { + // cleanup + batchSizeSetting = Settings.builder().putNull(setting1.getKey()).build(); + client().admin().cluster().prepareUpdateSettings().setPersistentSettings(batchSizeSetting).execute().actionGet(); + } + } + + public void testBatchSizeMaxValue() { + Setting setting1 = GATEWAY_ALLOCATOR_BATCH_SIZE; + Settings batchSizeSetting = Settings.builder().put(setting1.getKey(), "11000").build(); + + assertThrows( + "failed to parse value [11000] for setting [" + setting1.getKey() + "], must be <= [10000]", + IllegalArgumentException.class, + () -> { + client().admin().cluster().prepareUpdateSettings().setPersistentSettings(batchSizeSetting).execute().actionGet(); + } + ); + } + + public void testBatchSizeMinValue() { + Setting setting1 = GATEWAY_ALLOCATOR_BATCH_SIZE; + Settings batchSizeSetting = Settings.builder().put(setting1.getKey(), "0").build(); + + assertThrows( + "failed to parse value [0] for setting [" + setting1.getKey() + "], must be >= [1]", + IllegalArgumentException.class, + () -> { + client().admin().cluster().prepareUpdateSettings().setPersistentSettings(batchSizeSetting).execute().actionGet(); + } + ); + } +} diff --git a/server/src/test/java/org/opensearch/indices/replication/SegmentReplicatorTests.java b/server/src/test/java/org/opensearch/indices/replication/SegmentReplicatorTests.java index 7acee449a1b46..81ea16c80dd79 100644 --- a/server/src/test/java/org/opensearch/indices/replication/SegmentReplicatorTests.java +++ b/server/src/test/java/org/opensearch/indices/replication/SegmentReplicatorTests.java @@ -10,7 +10,13 @@ import org.apache.lucene.store.IOContext; import org.opensearch.OpenSearchCorruptionException; +import org.opensearch.cluster.ClusterState; import org.opensearch.cluster.metadata.IndexMetadata; +import org.opensearch.cluster.routing.IndexRoutingTable; +import org.opensearch.cluster.routing.IndexShardRoutingTable; +import org.opensearch.cluster.routing.RoutingTable; +import org.opensearch.cluster.routing.UnassignedInfo; +import org.opensearch.cluster.service.ClusterService; import org.opensearch.common.lucene.Lucene; import org.opensearch.common.settings.Settings; import org.opensearch.core.action.ActionListener; @@ -19,10 +25,12 @@ import org.opensearch.index.shard.IndexShard; import org.opensearch.index.shard.IndexShardTestCase; import org.opensearch.index.store.StoreFileMetadata; +import org.opensearch.indices.recovery.RecoverySettings; import org.opensearch.indices.replication.checkpoint.ReplicationCheckpoint; import org.opensearch.indices.replication.common.CopyState; import org.opensearch.indices.replication.common.ReplicationType; import org.opensearch.threadpool.ThreadPool; +import org.opensearch.transport.TransportService; import java.io.IOException; import java.io.UncheckedIOException; @@ -45,6 +53,48 @@ public class SegmentReplicatorTests extends IndexShardTestCase { .put(IndexMetadata.SETTING_REPLICATION_TYPE, ReplicationType.SEGMENT) .build(); + public void testReplicationWithUnassignedPrimary() throws Exception { + final IndexShard replica = newStartedShard(false, settings, new NRTReplicationEngineFactory()); + final IndexShard primary = newStartedShard(true, settings, new NRTReplicationEngineFactory()); + SegmentReplicator replicator = new SegmentReplicator(threadPool); + + ClusterService cs = mock(ClusterService.class); + IndexShardRoutingTable.Builder shardRoutingTable = new IndexShardRoutingTable.Builder(replica.shardId()); + shardRoutingTable.addShard(replica.routingEntry()); + shardRoutingTable.addShard(primary.routingEntry().moveToUnassigned(new UnassignedInfo(UnassignedInfo.Reason.NODE_LEFT, "test"))); + + when(cs.state()).thenReturn(buildClusterState(replica, shardRoutingTable)); + replicator.setSourceFactory(new SegmentReplicationSourceFactory(mock(TransportService.class), mock(RecoverySettings.class), cs)); + expectThrows(IllegalStateException.class, () -> replicator.startReplication(replica)); + closeShards(replica, primary); + } + + public void testReplicationWithUnknownPrimaryNode() throws Exception { + final IndexShard replica = newStartedShard(false, settings, new NRTReplicationEngineFactory()); + final IndexShard primary = newStartedShard(true, settings, new NRTReplicationEngineFactory()); + SegmentReplicator replicator = new SegmentReplicator(threadPool); + + ClusterService cs = mock(ClusterService.class); + IndexShardRoutingTable.Builder shardRoutingTable = new IndexShardRoutingTable.Builder(replica.shardId()); + shardRoutingTable.addShard(replica.routingEntry()); + shardRoutingTable.addShard(primary.routingEntry()); + + when(cs.state()).thenReturn(buildClusterState(replica, shardRoutingTable)); + replicator.setSourceFactory(new SegmentReplicationSourceFactory(mock(TransportService.class), mock(RecoverySettings.class), cs)); + expectThrows(IllegalStateException.class, () -> replicator.startReplication(replica)); + closeShards(replica, primary); + } + + private ClusterState buildClusterState(IndexShard replica, IndexShardRoutingTable.Builder indexShard) { + return ClusterState.builder(clusterService.state()) + .routingTable( + RoutingTable.builder() + .add(IndexRoutingTable.builder(replica.shardId().getIndex()).addIndexShard(indexShard.build()).build()) + .build() + ) + .build(); + } + public void testStartReplicationWithoutSourceFactory() { ThreadPool threadpool = mock(ThreadPool.class); ExecutorService mock = mock(ExecutorService.class); diff --git a/server/src/test/java/org/opensearch/rest/action/admin/cluster/RestClusterStatsActionTests.java b/server/src/test/java/org/opensearch/rest/action/admin/cluster/RestClusterStatsActionTests.java new file mode 100644 index 0000000000000..8b46f676636fd --- /dev/null +++ b/server/src/test/java/org/opensearch/rest/action/admin/cluster/RestClusterStatsActionTests.java @@ -0,0 +1,171 @@ +/* + * SPDX-License-Identifier: Apache-2.0 + * + * The OpenSearch Contributors require contributions made to + * this file be licensed under the Apache-2.0 license or a + * compatible open source license. + */ + +package org.opensearch.rest.action.admin.cluster; + +import org.opensearch.action.admin.cluster.stats.ClusterStatsRequest; +import org.opensearch.client.node.NodeClient; +import org.opensearch.rest.RestRequest; +import org.opensearch.test.OpenSearchTestCase; +import org.opensearch.test.rest.FakeRestRequest; + +import java.util.HashMap; +import java.util.Set; +import java.util.stream.Collectors; + +import static org.hamcrest.CoreMatchers.containsString; +import static org.hamcrest.object.HasToString.hasToString; +import static org.mockito.Mockito.mock; + +public class RestClusterStatsActionTests extends OpenSearchTestCase { + + private RestClusterStatsAction action; + + @Override + public void setUp() throws Exception { + super.setUp(); + action = new RestClusterStatsAction(); + } + + public void testFromRequestBasePath() { + final HashMap params = new HashMap<>(); + final RestRequest request = new FakeRestRequest.Builder(xContentRegistry()).withPath("/_cluster/stats").withParams(params).build(); + ClusterStatsRequest clusterStatsRequest = RestClusterStatsAction.fromRequest(request); + assertNotNull(clusterStatsRequest); + assertTrue(clusterStatsRequest.useAggregatedNodeLevelResponses()); + assertFalse(clusterStatsRequest.computeAllMetrics()); + assertNotNull(clusterStatsRequest.requestedMetrics()); + assertFalse(clusterStatsRequest.requestedMetrics().isEmpty()); + for (ClusterStatsRequest.Metric metric : ClusterStatsRequest.Metric.values()) { + assertTrue(clusterStatsRequest.requestedMetrics().contains(metric)); + } + assertNotNull(clusterStatsRequest.indicesMetrics()); + assertFalse(clusterStatsRequest.indicesMetrics().isEmpty()); + for (ClusterStatsRequest.IndexMetric indexMetric : ClusterStatsRequest.IndexMetric.values()) { + assertTrue(clusterStatsRequest.indicesMetrics().contains(indexMetric)); + } + } + + public void testFromRequestWithNodeStatsMetricsFilter() { + Set metricsRequested = Set.of(ClusterStatsRequest.Metric.OS, ClusterStatsRequest.Metric.FS); + String metricParam = metricsRequested.stream().map(ClusterStatsRequest.Metric::metricName).collect(Collectors.joining(",")); + final HashMap params = new HashMap<>(); + params.put("metric", metricParam); + final RestRequest request = new FakeRestRequest.Builder(xContentRegistry()).withPath("/_cluster/stats").withParams(params).build(); + ClusterStatsRequest clusterStatsRequest = RestClusterStatsAction.fromRequest(request); + assertNotNull(clusterStatsRequest); + assertTrue(clusterStatsRequest.useAggregatedNodeLevelResponses()); + assertFalse(clusterStatsRequest.computeAllMetrics()); + assertFalse(clusterStatsRequest.requestedMetrics().isEmpty()); + assertEquals(2, clusterStatsRequest.requestedMetrics().size()); + assertEquals(metricsRequested, clusterStatsRequest.requestedMetrics()); + assertTrue(clusterStatsRequest.indicesMetrics().isEmpty()); + } + + public void testFromRequestWithIndicesStatsMetricsFilter() { + Set metricsRequested = Set.of( + ClusterStatsRequest.Metric.OS, + ClusterStatsRequest.Metric.FS, + ClusterStatsRequest.Metric.INDICES + ); + Set indicesMetricsRequested = Set.of( + ClusterStatsRequest.IndexMetric.SHARDS, + ClusterStatsRequest.IndexMetric.SEGMENTS + ); + String metricParam = metricsRequested.stream().map(ClusterStatsRequest.Metric::metricName).collect(Collectors.joining(",")); + String indicesMetricParam = indicesMetricsRequested.stream() + .map(ClusterStatsRequest.IndexMetric::metricName) + .collect(Collectors.joining(",")); + final HashMap params = new HashMap<>(); + params.put("metric", metricParam); + params.put("index_metric", indicesMetricParam); + final RestRequest request = new FakeRestRequest.Builder(xContentRegistry()).withPath("/_cluster/stats").withParams(params).build(); + ClusterStatsRequest clusterStatsRequest = RestClusterStatsAction.fromRequest(request); + assertNotNull(clusterStatsRequest); + assertTrue(clusterStatsRequest.useAggregatedNodeLevelResponses()); + assertFalse(clusterStatsRequest.computeAllMetrics()); + assertFalse(clusterStatsRequest.requestedMetrics().isEmpty()); + assertEquals(3, clusterStatsRequest.requestedMetrics().size()); + assertEquals(metricsRequested, clusterStatsRequest.requestedMetrics()); + assertFalse(clusterStatsRequest.indicesMetrics().isEmpty()); + assertEquals(2, clusterStatsRequest.indicesMetrics().size()); + assertEquals(indicesMetricsRequested, clusterStatsRequest.indicesMetrics()); + } + + public void testUnrecognizedMetric() { + final HashMap params = new HashMap<>(); + final String metric = randomAlphaOfLength(64); + params.put("metric", metric); + final RestRequest request = new FakeRestRequest.Builder(xContentRegistry()).withPath("/_cluster/stats").withParams(params).build(); + final IllegalArgumentException e = expectThrows( + IllegalArgumentException.class, + () -> action.prepareRequest(request, mock(NodeClient.class)) + ); + assertThat(e, hasToString(containsString("request [/_cluster/stats] contains unrecognized metric: [" + metric + "]"))); + } + + public void testUnrecognizedIndexMetric() { + final HashMap params = new HashMap<>(); + params.put("metric", "_all,"); + final String indexMetric = randomAlphaOfLength(64); + params.put("index_metric", indexMetric); + final RestRequest request = new FakeRestRequest.Builder(xContentRegistry()).withPath("/_cluster/stats").withParams(params).build(); + final IllegalArgumentException e = expectThrows( + IllegalArgumentException.class, + () -> action.prepareRequest(request, mock(NodeClient.class)) + ); + assertThat(e, hasToString(containsString("request [/_cluster/stats] contains unrecognized index metric: [" + indexMetric + "]"))); + } + + public void testAllMetricsRequestWithOtherMetric() { + final HashMap params = new HashMap<>(); + final String metric = randomSubsetOf(1, RestClusterStatsAction.METRIC_REQUEST_CONSUMER_MAP.keySet()).get(0); + params.put("metric", "_all," + metric); + final RestRequest request = new FakeRestRequest.Builder(xContentRegistry()).withPath("/_cluster/stats").withParams(params).build(); + final IllegalArgumentException e = expectThrows( + IllegalArgumentException.class, + () -> action.prepareRequest(request, mock(NodeClient.class)) + ); + assertThat(e, hasToString(containsString("request [/_cluster/stats] contains _all and individual metrics [_all," + metric + "]"))); + } + + public void testAllIndexMetricsRequestWithOtherIndicesMetric() { + final HashMap params = new HashMap<>(); + params.put("metric", "_all,"); + final String indexMetric = randomSubsetOf(1, RestClusterStatsAction.INDEX_METRIC_TO_REQUEST_CONSUMER_MAP.keySet()).get(0); + params.put("index_metric", "_all," + indexMetric); + final RestRequest request = new FakeRestRequest.Builder(xContentRegistry()).withPath("/_cluster/stats").withParams(params).build(); + final IllegalArgumentException e = expectThrows( + IllegalArgumentException.class, + () -> action.prepareRequest(request, mock(NodeClient.class)) + ); + assertThat( + e, + hasToString(containsString("request [/_cluster/stats] contains _all and individual index metrics [_all," + indexMetric + "]")) + ); + } + + public void testIndexMetricsRequestWithoutMetricIndices() { + final HashMap params = new HashMap<>(); + params.put("metric", "os"); + final String indexMetric = randomSubsetOf(1, RestClusterStatsAction.INDEX_METRIC_TO_REQUEST_CONSUMER_MAP.keySet()).get(0); + params.put("index_metric", indexMetric); + final RestRequest request = new FakeRestRequest.Builder(xContentRegistry()).withPath("/_cluster/stats").withParams(params).build(); + final IllegalArgumentException e = expectThrows( + IllegalArgumentException.class, + () -> action.prepareRequest(request, mock(NodeClient.class)) + ); + assertThat( + e, + hasToString( + containsString("request [/_cluster/stats] contains index metrics [" + indexMetric + "] but indices stats not requested") + ) + ); + } + +} diff --git a/server/src/test/java/org/opensearch/wlm/QueryGroupServiceTests.java b/server/src/test/java/org/opensearch/wlm/QueryGroupServiceTests.java index 45428865259c3..f22759ce968aa 100644 --- a/server/src/test/java/org/opensearch/wlm/QueryGroupServiceTests.java +++ b/server/src/test/java/org/opensearch/wlm/QueryGroupServiceTests.java @@ -48,6 +48,7 @@ import static org.mockito.Mockito.when; public class QueryGroupServiceTests extends OpenSearchTestCase { + public static final String QUERY_GROUP_ID = "queryGroupId1"; private QueryGroupService queryGroupService; private QueryGroupTaskCancellationService mockCancellationService; private ClusterService mockClusterService; @@ -68,6 +69,7 @@ public void setUp() throws Exception { mockNodeDuressTrackers = Mockito.mock(NodeDuressTrackers.class); mockCancellationService = Mockito.mock(TestQueryGroupCancellationService.class); mockQueryGroupsStateAccessor = new QueryGroupsStateAccessor(); + when(mockNodeDuressTrackers.isNodeInDuress()).thenReturn(false); queryGroupService = new QueryGroupService( mockCancellationService, @@ -203,26 +205,52 @@ public void testRejectIfNeeded_whenQueryGroupIdIsNullOrDefaultOne() { verify(spyMap, never()).get(any()); } + public void testRejectIfNeeded_whenSoftModeQueryGroupIsContendedAndNodeInDuress() { + Set activeQueryGroups = getActiveQueryGroups( + "testQueryGroup", + QUERY_GROUP_ID, + MutableQueryGroupFragment.ResiliencyMode.SOFT, + Map.of(ResourceType.CPU, 0.10) + ); + mockQueryGroupStateMap = new HashMap<>(); + mockQueryGroupStateMap.put("queryGroupId1", new QueryGroupState()); + QueryGroupState state = new QueryGroupState(); + QueryGroupState.ResourceTypeState cpuResourceState = new QueryGroupState.ResourceTypeState(ResourceType.CPU); + cpuResourceState.setLastRecordedUsage(0.10); + state.getResourceState().put(ResourceType.CPU, cpuResourceState); + QueryGroupState spyState = spy(state); + mockQueryGroupStateMap.put(QUERY_GROUP_ID, spyState); + + mockQueryGroupsStateAccessor = new QueryGroupsStateAccessor(mockQueryGroupStateMap); + + queryGroupService = new QueryGroupService( + mockCancellationService, + mockClusterService, + mockThreadPool, + mockWorkloadManagementSettings, + mockNodeDuressTrackers, + mockQueryGroupsStateAccessor, + activeQueryGroups, + new HashSet<>() + ); + when(mockWorkloadManagementSettings.getWlmMode()).thenReturn(WlmMode.ENABLED); + when(mockNodeDuressTrackers.isNodeInDuress()).thenReturn(true); + assertThrows(OpenSearchRejectedExecutionException.class, () -> queryGroupService.rejectIfNeeded("queryGroupId1")); + } + public void testRejectIfNeeded_whenQueryGroupIsSoftMode() { - QueryGroup testQueryGroup = new QueryGroup( + Set activeQueryGroups = getActiveQueryGroups( "testQueryGroup", - "queryGroupId1", - new MutableQueryGroupFragment(MutableQueryGroupFragment.ResiliencyMode.SOFT, Map.of(ResourceType.CPU, 0.10)), - 1L + QUERY_GROUP_ID, + MutableQueryGroupFragment.ResiliencyMode.SOFT, + Map.of(ResourceType.CPU, 0.10) ); - Set activeQueryGroups = new HashSet<>() { - { - add(testQueryGroup); - } - }; mockQueryGroupStateMap = new HashMap<>(); QueryGroupState spyState = spy(new QueryGroupState()); mockQueryGroupStateMap.put("queryGroupId1", spyState); mockQueryGroupsStateAccessor = new QueryGroupsStateAccessor(mockQueryGroupStateMap); - Map spyMap = spy(mockQueryGroupStateMap); - queryGroupService = new QueryGroupService( mockCancellationService, mockClusterService, @@ -239,11 +267,11 @@ public void testRejectIfNeeded_whenQueryGroupIsSoftMode() { } public void testRejectIfNeeded_whenQueryGroupIsEnforcedMode_andNotBreaching() { - QueryGroup testQueryGroup = new QueryGroup( + QueryGroup testQueryGroup = getQueryGroup( "testQueryGroup", "queryGroupId1", - new MutableQueryGroupFragment(MutableQueryGroupFragment.ResiliencyMode.ENFORCED, Map.of(ResourceType.CPU, 0.10)), - 1L + MutableQueryGroupFragment.ResiliencyMode.ENFORCED, + Map.of(ResourceType.CPU, 0.10) ); QueryGroup spuQueryGroup = spy(testQueryGroup); Set activeQueryGroups = new HashSet<>() { @@ -464,6 +492,31 @@ public void testShouldSBPHandle() { } + private static Set getActiveQueryGroups( + String name, + String id, + MutableQueryGroupFragment.ResiliencyMode mode, + Map resourceLimits + ) { + QueryGroup testQueryGroup = getQueryGroup(name, id, mode, resourceLimits); + Set activeQueryGroups = new HashSet<>() { + { + add(testQueryGroup); + } + }; + return activeQueryGroups; + } + + private static QueryGroup getQueryGroup( + String name, + String id, + MutableQueryGroupFragment.ResiliencyMode mode, + Map resourceLimits + ) { + QueryGroup testQueryGroup = new QueryGroup(name, id, new MutableQueryGroupFragment(mode, resourceLimits), 1L); + return testQueryGroup; + } + // This is needed to test the behavior of QueryGroupService#doRun method static class TestQueryGroupCancellationService extends QueryGroupTaskCancellationService { public TestQueryGroupCancellationService(