-
Notifications
You must be signed in to change notification settings - Fork 1.9k
New issue
Have a question about this project? Sign up for a free GitHub account to open an issue and contact its maintainers and the community.
By clicking “Sign up for GitHub”, you agree to our terms of service and privacy statement. We’ll occasionally send you account related emails.
Already on GitHub? Sign in to your account
[Segment Replication] Implementing cat/segment_replication API #5718
Merged
mch2
merged 37 commits into
opensearch-project:main
from
Rishikesh1159:segment_replication_API
Feb 8, 2023
+1,530
−43
Merged
Changes from all commits
Commits
Show all changes
37 commits
Select commit
Hold shift + click to select a range
2eed8a8
Initial Draft for adding segment_replication API
Rishikesh1159 88cf749
Adding bytes transfered in each segrep events and additional metrics.
Rishikesh1159 888c9b3
Merge branch 'main' into segment_replication_API
Rishikesh1159 ba048a4
Fix broken tests.
Rishikesh1159 69bc392
Merge branch 'segment_replication_API' of https://github.com/Rishikes…
Rishikesh1159 b062420
Fix compile errors
Rishikesh1159 1728ae7
Adding Tests and gating logic behind feature flag.
Rishikesh1159 b10519e
Add java docs and enable query parameter detailed.
Rishikesh1159 74a41b0
Add temporary documentation URL
Rishikesh1159 7752a00
Fixing failing tests.
Rishikesh1159 594814e
Merge branch 'main' into segment_replication_API
Rishikesh1159 9e62cd4
Merge branch 'segment_replication_API' of https://github.com/Rishikes…
Rishikesh1159 b5432bd
Spotless Apply.
Rishikesh1159 2b92c81
Fix media type copile check.
Rishikesh1159 5c9f8a7
Revert previous changes and fix failing tests.
Rishikesh1159 e022a7e
Apply spotless check.
Rishikesh1159 8b26e2e
Refactoring call to segmentreplicationstate.
Rishikesh1159 dd4f5b1
spotless check
Rishikesh1159 0e52667
Merge branch 'opensearch-project:main' into segment_replication_API
Rishikesh1159 c85e38a
Changing invokation of segment replication shard and filtering API re…
Rishikesh1159 ad1fd5b
Merge branch 'segment_replication_API' of https://github.com/Rishikes…
Rishikesh1159 2ef7a9f
disable feature flag by default.
Rishikesh1159 9a4f09d
Apply spotless
Rishikesh1159 1080536
Address comments on PR.
Rishikesh1159 86cf0ae
Merge branch 'main' into segment_replication_API
Rishikesh1159 8ae10c8
Merge branch 'segment_replication_API' of https://github.com/Rishikes…
Rishikesh1159 9cf580e
Merge branch 'opensearch-project:main' into segment_replication_API
Rishikesh1159 b8f7b71
Fix gradle check failures
Rishikesh1159 08bc3d6
Merge branch 'segment_replication_API' of https://github.com/Rishikes…
Rishikesh1159 c1ef1d4
fix failing testSegment_ReplicationActionAction()
Rishikesh1159 884e6d1
Exclude empty segment replication events in API response.
Rishikesh1159 537964f
Apply spotless.
Rishikesh1159 f120a92
Address PR comments and add Integ Tests.
Rishikesh1159 35460e9
Merge branch 'opensearch-project:main' into segment_replication_API
Rishikesh1159 fc24661
Fix failing testSegmentReplicationApiResponse().
Rishikesh1159 f64f14c
Merge branch 'segment_replication_API' of https://github.com/Rishikes…
Rishikesh1159 7d07dc6
Refactoring code.
Rishikesh1159 File filter
Filter by extension
Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
There are no files selected for viewing
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
108 changes: 108 additions & 0 deletions
108
rest-api-spec/src/main/resources/rest-api-spec/api/cat.segment_replication.json
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,108 @@ | ||
{ | ||
"cat.segment_replication":{ | ||
"documentation":{ | ||
"url":"https://github.com/opensearch-project/documentation-website/issues/2627", | ||
"description":"Returns information about both on-going and latest completed Segment Replication events" | ||
}, | ||
"stability":"stable", | ||
"url":{ | ||
"paths":[ | ||
{ | ||
"path":"/_cat/segment_replication", | ||
"methods":[ | ||
"GET" | ||
] | ||
}, | ||
{ | ||
"path":"/_cat/segment_replication/{index}", | ||
"methods":[ | ||
"GET" | ||
], | ||
"parts":{ | ||
"index":{ | ||
"type":"list", | ||
"description":"Comma-separated list or wildcard expression of index names to limit the returned information" | ||
} | ||
} | ||
} | ||
] | ||
}, | ||
"params":{ | ||
"format":{ | ||
"type":"string", | ||
"description":"a short version of the Accept header, e.g. json, yaml" | ||
}, | ||
"active_only":{ | ||
There was a problem hiding this comment. Choose a reason for hiding this commentThe reason will be displayed to describe this comment to others. Learn more. nice |
||
"type":"boolean", | ||
"description":"If `true`, the response only includes ongoing segment replication events", | ||
"default":false | ||
}, | ||
"completed_only":{ | ||
"type":"boolean", | ||
"description":"If `true`, the response only includes latest completed segment replication events", | ||
"default":false | ||
}, | ||
"bytes":{ | ||
"type":"enum", | ||
"description":"The unit in which to display byte values", | ||
"options":[ | ||
"b", | ||
"k", | ||
"kb", | ||
"m", | ||
"mb", | ||
"g", | ||
"gb", | ||
"t", | ||
"tb", | ||
"p", | ||
"pb" | ||
] | ||
}, | ||
"detailed":{ | ||
"type":"boolean", | ||
"description":"If `true`, the response includes detailed information about segment replications", | ||
"default":false | ||
}, | ||
"shards":{ | ||
"type":"list", | ||
"description":"Comma-separated list of shards to display" | ||
}, | ||
"h":{ | ||
"type":"list", | ||
"description":"Comma-separated list of column names to display" | ||
}, | ||
"help":{ | ||
"type":"boolean", | ||
"description":"Return help information", | ||
"default":false | ||
}, | ||
"index":{ | ||
"type":"list", | ||
"description":"Comma-separated list or wildcard expression of index names to limit the returned information" | ||
}, | ||
"s":{ | ||
"type":"list", | ||
"description":"Comma-separated list of column names or column aliases to sort by" | ||
}, | ||
"time":{ | ||
"type":"enum", | ||
"description":"The unit in which to display time values", | ||
"options":[ | ||
"d", | ||
"h", | ||
"m", | ||
"s", | ||
"ms", | ||
"micros", | ||
"nanos" | ||
] | ||
}, | ||
"v":{ | ||
"type":"boolean", | ||
"description":"Verbose mode. Display column headers", | ||
"default":false | ||
} | ||
} | ||
} | ||
} |
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
154 changes: 154 additions & 0 deletions
154
...nternalClusterTest/java/org/opensearch/indices/replication/SegmentReplicationStatsIT.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,154 @@ | ||
/* | ||
* 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.indices.replication; | ||
|
||
import org.opensearch.OpenSearchStatusException; | ||
import org.opensearch.action.admin.indices.replication.SegmentReplicationStatsResponse; | ||
import org.opensearch.cluster.metadata.IndexMetadata; | ||
import org.opensearch.common.settings.Settings; | ||
import org.opensearch.indices.replication.common.ReplicationType; | ||
import org.opensearch.test.OpenSearchIntegTestCase; | ||
import org.opensearch.test.transport.MockTransportService; | ||
import org.opensearch.transport.TransportService; | ||
|
||
import java.util.concurrent.CountDownLatch; | ||
|
||
import static java.util.Arrays.asList; | ||
import static org.hamcrest.Matchers.equalTo; | ||
import static org.hamcrest.Matchers.greaterThan; | ||
|
||
@OpenSearchIntegTestCase.ClusterScope(scope = OpenSearchIntegTestCase.Scope.TEST, numDataNodes = 0) | ||
public class SegmentReplicationStatsIT extends SegmentReplicationBaseIT { | ||
|
||
public void testSegmentReplicationStatsResponse() throws Exception { | ||
final String primaryNode = internalCluster().startNode(); | ||
createIndex(INDEX_NAME); | ||
ensureYellowAndNoInitializingShards(INDEX_NAME); | ||
final String replicaNode = internalCluster().startNode(); | ||
ensureGreen(INDEX_NAME); | ||
|
||
// index 10 docs | ||
for (int i = 0; i < 10; i++) { | ||
client().prepareIndex(INDEX_NAME).setId(Integer.toString(i)).setSource("field", "value" + i).execute().actionGet(); | ||
} | ||
refresh(INDEX_NAME); | ||
waitForSearchableDocs(10L, asList(primaryNode, replicaNode)); | ||
|
||
SegmentReplicationStatsResponse response = client().admin() | ||
.indices() | ||
.prepareSegmentReplicationStats(INDEX_NAME) | ||
.execute() | ||
.actionGet(); | ||
// Verify API Response | ||
assertThat(response.shardSegmentReplicationStates().size(), equalTo(SHARD_COUNT)); | ||
assertThat(response.shardSegmentReplicationStates().get(INDEX_NAME).get(0).getStage(), equalTo(SegmentReplicationState.Stage.DONE)); | ||
assertThat(response.shardSegmentReplicationStates().get(INDEX_NAME).get(0).getIndex().recoveredFileCount(), greaterThan(0)); | ||
} | ||
|
||
public void testSegmentReplicationStatsResponseForActiveAndCompletedOnly() throws Exception { | ||
final String primaryNode = internalCluster().startNode(); | ||
createIndex(INDEX_NAME); | ||
ensureYellowAndNoInitializingShards(INDEX_NAME); | ||
final String replicaNode = internalCluster().startNode(); | ||
ensureGreen(INDEX_NAME); | ||
|
||
// index 10 docs | ||
for (int i = 0; i < 10; i++) { | ||
client().prepareIndex(INDEX_NAME).setId(Integer.toString(i)).setSource("field", "value" + i).execute().actionGet(); | ||
} | ||
refresh(INDEX_NAME); | ||
|
||
// index 10 more docs | ||
waitForSearchableDocs(10L, asList(primaryNode, replicaNode)); | ||
for (int i = 10; i < 20; i++) { | ||
client().prepareIndex(INDEX_NAME).setId(Integer.toString(i)).setSource("field", "value" + i).execute().actionGet(); | ||
} | ||
final CountDownLatch waitForReplication = new CountDownLatch(1); | ||
|
||
final CountDownLatch waitForAssertions = new CountDownLatch(1); | ||
// Mock transport service to add behaviour of waiting in GET_SEGMENT_FILES Stage of a segment replication event. | ||
MockTransportService mockTransportService = ((MockTransportService) internalCluster().getInstance( | ||
TransportService.class, | ||
replicaNode | ||
)); | ||
mockTransportService.addSendBehavior( | ||
internalCluster().getInstance(TransportService.class, primaryNode), | ||
(connection, requestId, action, request, options) -> { | ||
if (action.equals(SegmentReplicationSourceService.Actions.GET_SEGMENT_FILES)) { | ||
waitForReplication.countDown(); | ||
try { | ||
waitForAssertions.await(); | ||
} catch (InterruptedException e) { | ||
throw new RuntimeException(e); | ||
} | ||
} | ||
connection.sendRequest(requestId, action, request, options); | ||
} | ||
); | ||
refresh(INDEX_NAME); | ||
try { | ||
waitForReplication.await(); | ||
} catch (InterruptedException e) { | ||
throw new RuntimeException(e); | ||
} | ||
|
||
// verifying active_only by checking if current stage is GET_FILES STAGE | ||
SegmentReplicationStatsResponse activeOnlyResponse = client().admin() | ||
.indices() | ||
.prepareSegmentReplicationStats(INDEX_NAME) | ||
.setActiveOnly(true) | ||
.execute() | ||
.actionGet(); | ||
assertThat( | ||
activeOnlyResponse.shardSegmentReplicationStates().get(INDEX_NAME).get(0).getStage(), | ||
equalTo(SegmentReplicationState.Stage.GET_FILES) | ||
); | ||
|
||
// verifying completed_only by checking if current stage is DONE | ||
SegmentReplicationStatsResponse completedOnlyResponse = client().admin() | ||
.indices() | ||
.prepareSegmentReplicationStats(INDEX_NAME) | ||
.setCompletedOnly(true) | ||
.execute() | ||
.actionGet(); | ||
assertThat(completedOnlyResponse.shardSegmentReplicationStates().size(), equalTo(SHARD_COUNT)); | ||
assertThat( | ||
completedOnlyResponse.shardSegmentReplicationStates().get(INDEX_NAME).get(0).getStage(), | ||
equalTo(SegmentReplicationState.Stage.DONE) | ||
); | ||
waitForAssertions.countDown(); | ||
} | ||
|
||
public void testSegmentReplicationStatsResponseOnDocumentReplicationIndex() { | ||
final String primaryNode = internalCluster().startNode(); | ||
prepareCreate( | ||
INDEX_NAME, | ||
Settings.builder().put(IndexMetadata.SETTING_REPLICATION_TYPE, ReplicationType.DOCUMENT) | ||
|
||
).get(); | ||
ensureYellowAndNoInitializingShards(INDEX_NAME); | ||
final String replicaNode = internalCluster().startNode(); | ||
ensureGreen(INDEX_NAME); | ||
|
||
// index 10 docs | ||
for (int i = 0; i < 10; i++) { | ||
client().prepareIndex(INDEX_NAME).setId(Integer.toString(i)).setSource("field", "value" + i).execute().actionGet(); | ||
} | ||
refresh(INDEX_NAME); | ||
OpenSearchStatusException exception = assertThrows( | ||
OpenSearchStatusException.class, | ||
() -> client().admin().indices().prepareSegmentReplicationStats(INDEX_NAME).execute().actionGet() | ||
); | ||
// Verify exception message | ||
String expectedMessage = "Segment Replication is not enabled on Index: test-idx-1"; | ||
assertEquals(expectedMessage, exception.getMessage()); | ||
|
||
} | ||
|
||
} |
Oops, something went wrong.
Add this suggestion to a batch that can be applied as a single commit.
This suggestion is invalid because no changes were made to the code.
Suggestions cannot be applied while the pull request is closed.
Suggestions cannot be applied while viewing a subset of changes.
Only one suggestion per line can be applied in a batch.
Add this suggestion to a batch that can be applied as a single commit.
Applying suggestions on deleted lines is not supported.
You must change the existing code in this line in order to create a valid suggestion.
Outdated suggestions cannot be applied.
This suggestion has been applied or marked resolved.
Suggestions cannot be applied from pending reviews.
Suggestions cannot be applied on multi-line comments.
Suggestions cannot be applied while the pull request is queued to merge.
Suggestion cannot be applied right now. Please check back later.
There was a problem hiding this comment.
Choose a reason for hiding this comment
The reason will be displayed to describe this comment to others. Learn more.
@Rishikesh1159 Should probably make this
experimental
, no?There was a problem hiding this comment.
Choose a reason for hiding this comment
The reason will be displayed to describe this comment to others. Learn more.
Thanks @andrross for catching this. Yes I missed this, it should be
experimental
. Let me make a PR to change this.