Skip to content

Commit 722d6dd

Browse files
author
sukriti sinha
committed
Created an API to fetch remote store segment data
Signed-off-by: sukriti sinha <[email protected]>
1 parent 58c281f commit 722d6dd

18 files changed

+1531
-0
lines changed

‎server/src/internalClusterTest/java/org/opensearch/remotestore/RemoteStoreMetadataIT.java

Lines changed: 173 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -0,0 +1,173 @@
1+
/*
2+
* SPDX-License-Identifier: Apache-2.0
3+
*
4+
* The OpenSearch Contributors require contributions made to
5+
* this file be licensed under the Apache-2.0 license or a
6+
* compatible open source license.
7+
*/
8+
9+
package org.opensearch.remotestore;
10+
11+
import org.opensearch.action.admin.cluster.remotestore.metadata.RemoteStoreMetadata;
12+
import org.opensearch.action.admin.cluster.remotestore.metadata.RemoteStoreMetadataResponse;
13+
import org.opensearch.cluster.ClusterState;
14+
import org.opensearch.cluster.node.DiscoveryNode;
15+
import org.opensearch.plugins.Plugin;
16+
import org.opensearch.test.OpenSearchIntegTestCase;
17+
import org.opensearch.test.transport.MockTransportService;
18+
19+
import java.util.Collection;
20+
import java.util.List;
21+
import java.util.Map;
22+
import java.util.stream.Collectors;
23+
import java.util.stream.Stream;
24+
25+
import static org.hamcrest.Matchers.allOf;
26+
import static org.hamcrest.Matchers.hasKey;
27+
28+
@OpenSearchIntegTestCase.ClusterScope(scope = OpenSearchIntegTestCase.Scope.TEST, numDataNodes = 0)
29+
public class RemoteStoreMetadataIT extends RemoteStoreBaseIntegTestCase {
30+
31+
private static final String INDEX_NAME = "remote-store-meta-api-test";
32+
33+
@Override
34+
protected Collection<Class<? extends Plugin>> nodePlugins() {
35+
return Stream.concat(super.nodePlugins().stream(), Stream.of(MockTransportService.TestPlugin.class)).collect(Collectors.toList());
36+
}
37+
38+
public void setup() {
39+
internalCluster().startNodes(3);
40+
}
41+
42+
@SuppressWarnings("unchecked")
43+
public void testMetadataResponseFromAllNodes() {
44+
setup();
45+
46+
createIndex(INDEX_NAME, remoteStoreIndexSettings(0, 3));
47+
ensureGreen(INDEX_NAME);
48+
indexDocs();
49+
client().admin().indices().prepareRefresh(INDEX_NAME).get();
50+
51+
ClusterState state = getClusterState();
52+
List<String> nodes = state.nodes().getNodes().values().stream().map(DiscoveryNode::getName).collect(Collectors.toList());
53+
54+
for (String node : nodes) {
55+
RemoteStoreMetadataResponse response = client(node).admin().cluster().prepareRemoteStoreMetadata(INDEX_NAME, null).get();
56+
assertTrue(response.getSuccessfulShards() > 0);
57+
assertNotNull(response.groupByIndexAndShards());
58+
59+
response.groupByIndexAndShards().forEach((index, shardMap) -> {
60+
shardMap.forEach((shardId, metadataList) -> {
61+
assertFalse(metadataList.isEmpty());
62+
63+
for (RemoteStoreMetadata metadata : metadataList) {
64+
assertEquals(index, metadata.getIndexName());
65+
assertEquals((int) shardId, metadata.getShardId());
66+
67+
Map<String, Object> segments = metadata.getSegments();
68+
assertNotNull(segments);
69+
assertTrue(segments.containsKey("files"));
70+
Map<String, Object> files = (Map<String, Object>) segments.get("files");
71+
assertFalse(files.isEmpty());
72+
73+
for (Map.Entry<String, Object> entry : files.entrySet()) {
74+
Map<String, Object> fileMeta = (Map<String, Object>) entry.getValue();
75+
assertThat(fileMeta, allOf(hasKey("original_name"), hasKey("checksum"), hasKey("length")));
76+
}
77+
78+
assertTrue(segments.containsKey("replication_checkpoint"));
79+
Map<String, Object> checkpoint = (Map<String, Object>) segments.get("replication_checkpoint");
80+
assertThat(
81+
checkpoint,
82+
allOf(
83+
hasKey("primary_term"),
84+
hasKey("segments_gen"),
85+
hasKey("segment_infos_version"),
86+
hasKey("codec"),
87+
hasKey("created_timestamp")
88+
)
89+
);
90+
91+
Map<String, Object> translog = metadata.getTranslog();
92+
assertNotNull(translog);
93+
assertThat(
94+
translog,
95+
allOf(
96+
hasKey("primary_term"),
97+
hasKey("generation"),
98+
hasKey("min_translog_gen"),
99+
hasKey("generation_to_primary_term")
100+
)
101+
);
102+
}
103+
});
104+
});
105+
}
106+
}
107+
108+
@SuppressWarnings("unchecked")
109+
public void testMetadataResponseAllShards() {
110+
setup();
111+
112+
createIndex(INDEX_NAME, remoteStoreIndexSettings(0, 2));
113+
ensureGreen(INDEX_NAME);
114+
indexDocs();
115+
client().admin().indices().prepareRefresh(INDEX_NAME).get();
116+
117+
RemoteStoreMetadataResponse response = client().admin().cluster().prepareRemoteStoreMetadata(INDEX_NAME, null).get();
118+
assertEquals(2, response.getSuccessfulShards());
119+
120+
response.groupByIndexAndShards().forEach((index, shardMap) -> {
121+
shardMap.forEach((shardId, metadataList) -> {
122+
assertFalse(metadataList.isEmpty());
123+
124+
for (RemoteStoreMetadata metadata : metadataList) {
125+
assertEquals(index, metadata.getIndexName());
126+
assertEquals((int) shardId, metadata.getShardId());
127+
128+
Map<String, Object> segments = metadata.getSegments();
129+
assertNotNull(segments);
130+
assertTrue(segments.containsKey("files"));
131+
Map<String, Object> files = (Map<String, Object>) segments.get("files");
132+
assertFalse(files.isEmpty());
133+
134+
for (Map.Entry<String, Object> entry : files.entrySet()) {
135+
Map<String, Object> fileMeta = (Map<String, Object>) entry.getValue();
136+
assertThat(fileMeta, allOf(hasKey("original_name"), hasKey("checksum"), hasKey("length")));
137+
}
138+
139+
assertTrue(segments.containsKey("replication_checkpoint"));
140+
Map<String, Object> checkpoint = (Map<String, Object>) segments.get("replication_checkpoint");
141+
assertThat(
142+
checkpoint,
143+
allOf(
144+
hasKey("primary_term"),
145+
hasKey("segments_gen"),
146+
hasKey("segment_infos_version"),
147+
hasKey("codec"),
148+
hasKey("created_timestamp")
149+
)
150+
);
151+
152+
Map<String, Object> translog = metadata.getTranslog();
153+
assertNotNull(translog);
154+
assertThat(
155+
translog,
156+
allOf(
157+
hasKey("primary_term"),
158+
hasKey("generation"),
159+
hasKey("min_translog_gen"),
160+
hasKey("generation_to_primary_term")
161+
)
162+
);
163+
}
164+
});
165+
});
166+
}
167+
168+
private void indexDocs() {
169+
for (int i = 0; i < randomIntBetween(10, 20); i++) {
170+
client().prepareIndex(INDEX_NAME).setId("doc-" + i).setSource("field", "value-" + i).get();
171+
}
172+
}
173+
}

‎server/src/main/java/org/opensearch/action/ActionModule.java

Lines changed: 5 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -67,6 +67,8 @@
6767
import org.opensearch.action.admin.cluster.node.usage.TransportNodesUsageAction;
6868
import org.opensearch.action.admin.cluster.remote.RemoteInfoAction;
6969
import org.opensearch.action.admin.cluster.remote.TransportRemoteInfoAction;
70+
import org.opensearch.action.admin.cluster.remotestore.metadata.RemoteStoreMetadataAction;
71+
import org.opensearch.action.admin.cluster.remotestore.metadata.TransportRemoteStoreMetadataAction;
7072
import org.opensearch.action.admin.cluster.remotestore.restore.RestoreRemoteStoreAction;
7173
import org.opensearch.action.admin.cluster.remotestore.restore.TransportRestoreRemoteStoreAction;
7274
import org.opensearch.action.admin.cluster.remotestore.stats.RemoteStoreStatsAction;
@@ -378,6 +380,7 @@
378380
import org.opensearch.rest.action.admin.cluster.RestPutStoredScriptAction;
379381
import org.opensearch.rest.action.admin.cluster.RestReloadSecureSettingsAction;
380382
import org.opensearch.rest.action.admin.cluster.RestRemoteClusterInfoAction;
383+
import org.opensearch.rest.action.admin.cluster.RestRemoteStoreMetadataAction;
381384
import org.opensearch.rest.action.admin.cluster.RestRemoteStoreStatsAction;
382385
import org.opensearch.rest.action.admin.cluster.RestRestoreRemoteStoreAction;
383386
import org.opensearch.rest.action.admin.cluster.RestRestoreSnapshotAction;
@@ -638,6 +641,7 @@ public <Request extends ActionRequest, Response extends ActionResponse> void reg
638641
actions.register(NodesStatsAction.INSTANCE, TransportNodesStatsAction.class);
639642
actions.register(WlmStatsAction.INSTANCE, TransportWlmStatsAction.class);
640643
actions.register(RemoteStoreStatsAction.INSTANCE, TransportRemoteStoreStatsAction.class);
644+
actions.register(RemoteStoreMetadataAction.INSTANCE, TransportRemoteStoreMetadataAction.class);
641645
actions.register(NodesUsageAction.INSTANCE, TransportNodesUsageAction.class);
642646
actions.register(NodesHotThreadsAction.INSTANCE, TransportNodesHotThreadsAction.class);
643647
actions.register(ListTasksAction.INSTANCE, TransportListTasksAction.class);
@@ -1053,6 +1057,7 @@ public void initRestHandlers(Supplier<DiscoveryNodes> nodesInCluster) {
10531057
registerHandler.accept(new RestGetDecommissionStateAction());
10541058
registerHandler.accept(new RestRemoteStoreStatsAction());
10551059
registerHandler.accept(new RestRestoreRemoteStoreAction());
1060+
registerHandler.accept(new RestRemoteStoreMetadataAction());
10561061

10571062
// pull-based ingestion API
10581063
registerHandler.accept(new RestPauseIngestionAction());

‎server/src/main/java/org/opensearch/action/admin/cluster/remotestore/metadata/RemoteStoreMetadata.java

Lines changed: 92 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -0,0 +1,92 @@
1+
/*
2+
* SPDX-License-Identifier: Apache-2.0
3+
*
4+
* The OpenSearch Contributors require contributions made to
5+
* this file be licensed under the Apache-2.0 license or a
6+
* compatible open source license.
7+
*/
8+
9+
package org.opensearch.action.admin.cluster.remotestore.metadata;
10+
11+
import org.opensearch.common.annotation.ExperimentalApi;
12+
import org.opensearch.core.common.io.stream.StreamInput;
13+
import org.opensearch.core.common.io.stream.StreamOutput;
14+
import org.opensearch.core.common.io.stream.Writeable;
15+
import org.opensearch.core.xcontent.ToXContentFragment;
16+
import org.opensearch.core.xcontent.XContentBuilder;
17+
18+
import java.io.IOException;
19+
import java.util.Map;
20+
21+
/**
22+
* Response model that holds the remote store metadata (segment and translog) for a shard.
23+
*
24+
* @opensearch.internal
25+
*/
26+
@ExperimentalApi
27+
public class RemoteStoreMetadata implements Writeable, ToXContentFragment {
28+
private final Map<String, Object> segments;
29+
private final Map<String, Object> translog;
30+
private final String indexName;
31+
private final int shardId;
32+
33+
public RemoteStoreMetadata(Map<String, Object> segments, Map<String, Object> translog, String indexName, int shardId) {
34+
this.segments = segments;
35+
this.translog = translog;
36+
this.indexName = indexName;
37+
this.shardId = shardId;
38+
}
39+
40+
public RemoteStoreMetadata(StreamInput in) throws IOException {
41+
this.segments = in.readMap();
42+
this.translog = in.readMap();
43+
this.indexName = in.readString();
44+
this.shardId = in.readInt();
45+
}
46+
47+
@Override
48+
public void writeTo(StreamOutput out) throws IOException {
49+
out.writeMap(segments);
50+
out.writeMap(translog);
51+
out.writeString(indexName);
52+
out.writeInt(shardId);
53+
}
54+
55+
@Override
56+
public XContentBuilder toXContent(XContentBuilder builder, Params params) throws IOException {
57+
builder.startObject();
58+
59+
builder.field("index", indexName);
60+
builder.field("shard", shardId);
61+
62+
builder.startObject("segments");
63+
for (Map.Entry<String, Object> entry : segments.entrySet()) {
64+
builder.field(entry.getKey(), entry.getValue());
65+
}
66+
builder.endObject();
67+
68+
builder.startObject("translog");
69+
for (Map.Entry<String, Object> entry : translog.entrySet()) {
70+
builder.field(entry.getKey(), entry.getValue());
71+
}
72+
builder.endObject();
73+
74+
return builder.endObject();
75+
}
76+
77+
public String getIndexName() {
78+
return indexName;
79+
}
80+
81+
public int getShardId() {
82+
return shardId;
83+
}
84+
85+
public Map<String, Object> getSegments() {
86+
return segments;
87+
}
88+
89+
public Map<String, Object> getTranslog() {
90+
return translog;
91+
}
92+
}

‎server/src/main/java/org/opensearch/action/admin/cluster/remotestore/metadata/RemoteStoreMetadataAction.java

Lines changed: 25 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -0,0 +1,25 @@
1+
/*
2+
* SPDX-License-Identifier: Apache-2.0
3+
*
4+
* The OpenSearch Contributors require contributions made to
5+
* this file be licensed under the Apache-2.0 license or a
6+
* compatible open source license.
7+
*/
8+
9+
package org.opensearch.action.admin.cluster.remotestore.metadata;
10+
11+
import org.opensearch.action.ActionType;
12+
13+
/**
14+
* Action to fetch metadata from remote store
15+
*
16+
* @opensearch.internal
17+
*/
18+
public class RemoteStoreMetadataAction extends ActionType<RemoteStoreMetadataResponse> {
19+
public static final RemoteStoreMetadataAction INSTANCE = new RemoteStoreMetadataAction();
20+
public static final String NAME = "cluster:admin/remote_store/metadata";
21+
22+
private RemoteStoreMetadataAction() {
23+
super(NAME, RemoteStoreMetadataResponse::new);
24+
}
25+
}

‎server/src/main/java/org/opensearch/action/admin/cluster/remotestore/metadata/RemoteStoreMetadataRequest.java

Lines changed: 51 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -0,0 +1,51 @@
1+
/*
2+
* SPDX-License-Identifier: Apache-2.0
3+
*
4+
* The OpenSearch Contributors require contributions made to
5+
* this file be licensed under the Apache-2.0 license or a
6+
* compatible open source license.
7+
*/
8+
9+
package org.opensearch.action.admin.cluster.remotestore.metadata;
10+
11+
import org.opensearch.action.support.broadcast.BroadcastRequest;
12+
import org.opensearch.common.annotation.ExperimentalApi;
13+
import org.opensearch.core.common.io.stream.StreamInput;
14+
import org.opensearch.core.common.io.stream.StreamOutput;
15+
16+
import java.io.IOException;
17+
18+
/**
19+
* Request object for fetching remote store metadata of shards across one or more indices.
20+
*
21+
* @opensearch.internal
22+
*/
23+
@ExperimentalApi
24+
public class RemoteStoreMetadataRequest extends BroadcastRequest<RemoteStoreMetadataRequest> {
25+
private String[] shards;
26+
27+
public RemoteStoreMetadataRequest() {
28+
super((String[]) null);
29+
shards = new String[0];
30+
}
31+
32+
public RemoteStoreMetadataRequest(StreamInput in) throws IOException {
33+
super(in);
34+
shards = in.readStringArray();
35+
}
36+
37+
@Override
38+
public void writeTo(StreamOutput out) throws IOException {
39+
super.writeTo(out);
40+
out.writeStringArray(shards);
41+
}
42+
43+
public RemoteStoreMetadataRequest shards(String... shards) {
44+
this.shards = shards;
45+
return this;
46+
}
47+
48+
public String[] shards() {
49+
return this.shards;
50+
}
51+
}

0 commit comments

Comments
 (0)