From 19824275fe78014249143724c5b84a6211e6e642 Mon Sep 17 00:00:00 2001 From: SwethaGuptha <156877431+SwethaGuptha@users.noreply.github.com> Date: Tue, 22 Oct 2024 20:34:14 +0530 Subject: [PATCH 1/4] URI path filtering support in cluster stats API (#15938) * URI path filtering support in cluster stats API Signed-off-by: Swetha Guptha --- CHANGELOG.md | 3 +- .../opensearch/upgrades/ClusterStatsIT.java | 67 ++++ .../admin/cluster/stats/ClusterStatsIT.java | 376 ++++++++++++++++++ .../cluster/stats/ClusterStatsIndices.java | 109 +++-- .../cluster/stats/ClusterStatsNodes.java | 105 +++-- .../cluster/stats/ClusterStatsRequest.java | 142 +++++++ .../stats/ClusterStatsRequestBuilder.java | 20 + .../cluster/stats/ClusterStatsResponse.java | 51 ++- .../stats/TransportClusterStatsAction.java | 133 +++++-- .../index/cache/query/QueryCacheStats.java | 3 + .../index/fielddata/FieldDataStats.java | 3 + .../admin/cluster/RestClusterStatsAction.java | 135 ++++++- .../stats/ClusterStatsResponseTests.java | 281 +++++++++++++ .../cluster/RestClusterStatsActionTests.java | 171 ++++++++ 14 files changed, 1483 insertions(+), 116 deletions(-) create mode 100644 qa/rolling-upgrade/src/test/java/org/opensearch/upgrades/ClusterStatsIT.java create mode 100644 server/src/test/java/org/opensearch/action/admin/cluster/stats/ClusterStatsResponseTests.java create mode 100644 server/src/test/java/org/opensearch/rest/action/admin/cluster/RestClusterStatsActionTests.java diff --git a/CHANGELOG.md b/CHANGELOG.md index 76b65a6cd70dc..832871453028b 100644 --- a/CHANGELOG.md +++ b/CHANGELOG.md @@ -26,7 +26,8 @@ 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 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)) - [Star Tree - Search] Add support for metric aggregations with/without term query ([15289](https://github.com/opensearch-project/OpenSearch/pull/15289)) ### Dependencies 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/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/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/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/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/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") + ) + ); + } + +} From 20e233e3b432957fc46b9f1b6953c8b556c87997 Mon Sep 17 00:00:00 2001 From: gargharsh3134 <51459091+gargharsh3134@users.noreply.github.com> Date: Tue, 22 Oct 2024 22:28:58 +0530 Subject: [PATCH 2/4] Fixing inline javadocs usage in PaginationStrategy (#16428) Signed-off-by: Harsh Garg Co-authored-by: Harsh Garg --- .../action/pagination/PaginationStrategy.java | 13 +++++++++---- 1 file changed, 9 insertions(+), 4 deletions(-) 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()); From 267c68ef9d4e7df8559693350b4c013f493f402e Mon Sep 17 00:00:00 2001 From: Marc Handalian Date: Tue, 22 Oct 2024 10:52:34 -0700 Subject: [PATCH 3/4] Fix unclosed store references with node-node segrep when primary node is unknown. (#16106) This PR fixes a bug with node-node pull based replication where if the replica does not know the DiscoveryNode of its primary we would fail after constructing a SegmentReplicationTarget that holds a store reference. Only after replication is started would a failure occur because the source node is null, and the target would not get cleaned up. Push based replication already handled this case by catching any error and closing the target. This update ensures the validation is done before constructing our PrimaryShardReplicationSource, before any target object is created in both cases push and pull. Signed-off-by: Marc Handalian --- .../indices/settings/SearchOnlyReplicaIT.java | 2 + .../PrimaryShardReplicationSource.java | 2 + .../SegmentReplicationSourceFactory.java | 6 ++- .../replication/SegmentReplicatorTests.java | 50 +++++++++++++++++++ 4 files changed, 59 insertions(+), 1 deletion(-) 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/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/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); From 5120efbcd8bfceeb236421ddbe4c01f1069a1850 Mon Sep 17 00:00:00 2001 From: Andriy Redko Date: Tue, 22 Oct 2024 14:57:37 -0400 Subject: [PATCH 4/4] Update JDK to 23.0.1 (#16429) Signed-off-by: Andriy Redko --- .../java/org/opensearch/gradle/test/DistroTestPlugin.java | 4 ++-- buildSrc/version.properties | 2 +- 2 files changed, 3 insertions(+), 3 deletions(-) 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