-
Notifications
You must be signed in to change notification settings - Fork 1.9k
Commit
This commit does not belong to any branch on this repository, and may belong to a fork outside of the repository.
Compute Segment Replication stats for backpressure.
This PR introduces new mechanisms to keep track of the current replicas within a replication group intended to be used to apply backpressure. The new stats are also added to NodeStats. Signed-off-by: Marc Handalian <handalm@amazon.com>
- Loading branch information
Showing
26 changed files
with
754 additions
and
20 deletions.
There are no files selected for viewing
97 changes: 97 additions & 0 deletions
97
qa/smoke-test-http/src/test/java/org/opensearch/http/SegmentReplicationRestIT.java
This file contains bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
| Original file line number | Diff line number | Diff line change |
|---|---|---|
| @@ -0,0 +1,97 @@ | ||
| /* | ||
| * 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.http; | ||
|
|
||
| import org.apache.hc.core5.http.io.entity.EntityUtils; | ||
| import org.opensearch.client.Node; | ||
| import org.opensearch.client.Request; | ||
| import org.opensearch.client.Response; | ||
| import org.opensearch.common.settings.Settings; | ||
| import org.opensearch.common.util.FeatureFlags; | ||
| import org.opensearch.common.xcontent.XContentHelper; | ||
| import org.opensearch.common.xcontent.json.JsonXContent; | ||
| import org.opensearch.test.OpenSearchIntegTestCase; | ||
| import org.opensearch.test.XContentTestUtils; | ||
|
|
||
| import java.util.ArrayList; | ||
| import java.util.List; | ||
| import java.util.Map; | ||
|
|
||
| import static org.opensearch.rest.RestStatus.CREATED; | ||
| import static org.opensearch.rest.RestStatus.OK; | ||
| import static org.opensearch.test.hamcrest.OpenSearchAssertions.assertHitCount; | ||
|
|
||
| @OpenSearchIntegTestCase.ClusterScope(scope = OpenSearchIntegTestCase.Scope.SUITE, supportsDedicatedMasters = false, numDataNodes = 2, | ||
| numClientNodes = 0) | ||
| public class SegmentReplicationRestIT extends HttpSmokeTestCase { | ||
|
|
||
| @Override | ||
| protected Settings featureFlagSettings() { | ||
| return Settings.builder().put(super.featureFlagSettings()).put(FeatureFlags.REPLICATION_TYPE, "true").build(); | ||
| } | ||
|
|
||
| @Override | ||
| protected boolean addMockInternalEngine() { | ||
| return false; | ||
| } | ||
|
|
||
| @SuppressWarnings("unchecked") | ||
| public void testSegmentReplicationStats() throws Exception { | ||
| // create index. | ||
| Request createRequest = new Request("PUT", "/test_index"); | ||
| createRequest.setJsonEntity("{\"settings\": {\"index\": {\"number_of_shards\": 1, \"number_of_replicas\": 1, " + | ||
| "\"replication\": {\"type\": \"SEGMENT\"}}}}"); | ||
| final Response indexCreatedResponse = getRestClient().performRequest(createRequest); | ||
| assertEquals(indexCreatedResponse.getStatusLine().getStatusCode(), OK.getStatus()); | ||
| ensureGreen("test_index"); | ||
|
|
||
| //index a doc | ||
| Request successfulIndexingRequest = new Request("POST", "/test_index/_doc/"); | ||
| successfulIndexingRequest.setJsonEntity("{\"foo\": \"bar\"}"); | ||
| final Response indexSuccessFul = getRestClient().performRequest(successfulIndexingRequest); | ||
| assertEquals(indexSuccessFul.getStatusLine().getStatusCode(), CREATED.getStatus()); | ||
|
|
||
| assertBusy(() -> { | ||
| // wait for SR to run. | ||
| for (Node node : getRestClient().getNodes()) { | ||
| assertHitCount(client(node.getName()).prepareSearch("test_index").setSize(0).setPreference("_only_local").get(), 1); | ||
| } | ||
| }); | ||
| Request statsRequest = new Request("GET", "/_nodes/stats/segment_replication?pretty"); | ||
| final Response response = getRestClient().performRequest(statsRequest); | ||
| logger.info("Node stats response\n{}", EntityUtils.toString(response.getEntity())); | ||
| Map<String, Object> statsMap = XContentHelper.convertToMap(JsonXContent.jsonXContent, response.getEntity().getContent(), | ||
| true); | ||
| List<Object> nodes = new ArrayList<>(((Map<Object, Object>) statsMap.get("nodes")).values()); | ||
| assertEquals(2, nodes.size()); | ||
| XContentTestUtils.JsonMapView node1 = new XContentTestUtils.JsonMapView((Map<String, Object>) nodes.get(0)); | ||
| final Map<Object, Object> node1_map = node1.get("segment_replication"); | ||
| Map<Object, Object> primaryNode_map = node1_map; | ||
| if (node1_map.isEmpty()) { | ||
| XContentTestUtils.JsonMapView node2 = new XContentTestUtils.JsonMapView((Map<String, Object>) nodes.get(1)); | ||
| primaryNode_map = node2.get("segment_replication"); | ||
| } | ||
| List<Object> primary_values = new ArrayList<>(primaryNode_map | ||
| .values()); | ||
| assertEquals(1, primary_values.size()); | ||
| XContentTestUtils.JsonMapView shard1 = new XContentTestUtils.JsonMapView((Map<String, Object>) primary_values.get(0)); | ||
| Integer node1TotalLimitsRejections = shard1.get("rejected_requests"); | ||
| assertNotNull(node1TotalLimitsRejections); | ||
| String average_replication_lag = shard1.get("average_replication_lag"); | ||
| assertNotNull(average_replication_lag); | ||
| List<Object> shard1_replicas = new ArrayList<>(((Map<Object, Object>) shard1.get("replicas")).values()); | ||
| assertEquals(1, shard1_replicas.size()); | ||
| XContentTestUtils.JsonMapView replica = new XContentTestUtils.JsonMapView((Map<String, Object>) shard1_replicas.get(0)); | ||
| Integer checkpoints_behind = replica.get("checkpoints_behind"); | ||
| assertEquals(0, checkpoints_behind.intValue()); | ||
| assertNotNull(replica.get("bytes_behind")); | ||
| assertNotNull(replica.get("average_replication_lag")); | ||
| assertNotNull(replica.get("average_checkpoints_per_sync")); | ||
| } | ||
| } |
This file contains bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
This file contains bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
This file contains bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
This file contains bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
79 changes: 79 additions & 0 deletions
79
server/src/main/java/org/opensearch/index/SegmentReplicationPerGroupStats.java
This file contains bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
| Original file line number | Diff line number | Diff line change |
|---|---|---|
| @@ -0,0 +1,79 @@ | ||
| /* | ||
| * 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.index; | ||
|
|
||
| import org.opensearch.common.io.stream.StreamInput; | ||
| import org.opensearch.common.io.stream.StreamOutput; | ||
| import org.opensearch.common.io.stream.Writeable; | ||
| import org.opensearch.common.unit.TimeValue; | ||
| import org.opensearch.core.xcontent.ToXContentFragment; | ||
| import org.opensearch.core.xcontent.XContentBuilder; | ||
|
|
||
| import java.io.IOException; | ||
| import java.util.Set; | ||
| import java.util.concurrent.TimeUnit; | ||
|
|
||
| /** | ||
| * Return Segment Replication stats for a Replication Group. | ||
| * | ||
| * @opensearch.internal | ||
| */ | ||
| public class SegmentReplicationPerGroupStats implements Writeable, ToXContentFragment { | ||
|
|
||
| private final Set<SegmentReplicationShardStats> replicaStats; | ||
| private final long rejectedRequestCount; | ||
| private final TimeValue averageReplicationLag; | ||
|
|
||
| public SegmentReplicationPerGroupStats( | ||
| Set<SegmentReplicationShardStats> replicaStats, | ||
| long rejectedRequestCount, | ||
| TimeValue averageReplicationLag | ||
| ) { | ||
| this.replicaStats = replicaStats; | ||
| this.rejectedRequestCount = rejectedRequestCount; | ||
| this.averageReplicationLag = averageReplicationLag; | ||
| } | ||
|
|
||
| public SegmentReplicationPerGroupStats(StreamInput in) throws IOException { | ||
| this.replicaStats = in.readSet(SegmentReplicationShardStats::new); | ||
| this.rejectedRequestCount = in.readVLong(); | ||
| this.averageReplicationLag = new TimeValue(in.readVLong(), TimeUnit.MILLISECONDS); | ||
| } | ||
|
|
||
| public Set<SegmentReplicationShardStats> getReplicaStats() { | ||
| return replicaStats; | ||
| } | ||
|
|
||
| public long getRejectedRequestCount() { | ||
| return rejectedRequestCount; | ||
| } | ||
|
|
||
| public TimeValue getAverageReplicationLag() { | ||
| return averageReplicationLag; | ||
| } | ||
|
|
||
| @Override | ||
| public XContentBuilder toXContent(XContentBuilder builder, Params params) throws IOException { | ||
| builder.field("rejected_requests", rejectedRequestCount); | ||
| builder.field("average_replication_lag", averageReplicationLag); | ||
| builder.startObject("replicas"); | ||
| for (SegmentReplicationShardStats stats : replicaStats) { | ||
| stats.toXContent(builder, params); | ||
| } | ||
| builder.endObject(); | ||
| return builder; | ||
| } | ||
|
|
||
| @Override | ||
| public void writeTo(StreamOutput out) throws IOException { | ||
| out.writeCollection(replicaStats); | ||
| out.writeVLong(rejectedRequestCount); | ||
| out.writeVLong(averageReplicationLag.millis()); | ||
| } | ||
| } |
37 changes: 37 additions & 0 deletions
37
server/src/main/java/org/opensearch/index/SegmentReplicationPressureService.java
This file contains bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
| Original file line number | Diff line number | Diff line change |
|---|---|---|
| @@ -0,0 +1,37 @@ | ||
| /* | ||
| * 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.index; | ||
|
|
||
| import org.opensearch.action.admin.indices.stats.CommonStatsFlags; | ||
| import org.opensearch.index.shard.ShardId; | ||
| import org.opensearch.indices.IndicesService; | ||
|
|
||
| /** | ||
| * Service responsible for applying backpressure for lagging behind replicas when Segment Replication is enabled. | ||
| * | ||
| * @opensearch.internal | ||
| */ | ||
| public class SegmentReplicationPressureService { | ||
|
|
||
| private final IndicesService indicesService; | ||
| private final SegmentReplicationStatsTracker tracker; | ||
|
|
||
| public SegmentReplicationPressureService(IndicesService indexService) { | ||
| this.indicesService = indexService; | ||
| this.tracker = new SegmentReplicationStatsTracker(indicesService); | ||
| } | ||
|
|
||
| public void isSegrepLimitBreached(ShardId shardId) { | ||
| // TODO. | ||
| } | ||
|
|
||
| public SegmentReplicationStats nodeStats() { | ||
| return tracker.getStats(); | ||
| } | ||
| } |
Oops, something went wrong.