Skip to content

Commit

Permalink
Fix the support of RestClient Node Sniffer for version 2.x and update…
Browse files Browse the repository at this point in the history 
… tests (#3487)

Fix the support of RestClient Node Sniffer for OpenSearch 2.x, and update unit tests for OpenSearch.
The current code contains the logic to be compatible with Elasticsearch 2.x version, which is conflict with OpenSearch 2.x, so removed that part of legacy code.

* Update the script create_test_nodes_info.bash to dump the response of Nodes Info API GET _nodes/http for OpenSearch 1.0 and 2.0 version, which used for unit test.
* Remove the support of Elasticsearch version 2.x for the Sniffer
* Update unit test to validate the Sniffer compatible with OpenSearch 1.x and 2.x
* Update the API response parser to meet the array notation (in ES 6.1 and above) for the node attributes setting. It will result the value of `node.attr` setting will not be parsed as array in the Sniffer, when using the Sniffer on cluster in Elasticsearch 6.0 and above.
* Replace "master" node role with "cluster_manager" in unit test

Signed-off-by: Tianli Feng <ftianli@amazon.com>
  • Loading branch information
Tianli Feng committed Jun 6, 2022
1 parent 37b48ef commit ab478ba
Show file tree
Hide file tree
Showing 8 changed files with 191 additions and 700 deletions.
70 changes: 10 additions & 60 deletions client/sniffer/src/main/java/org/opensearch/client/sniff/OpenSearchNodesSniffer.java
View file
Original file line number Diff line number Diff line change
Expand Up @@ -49,6 +49,7 @@
import java.io.InputStream;
import java.net.URI;
import java.util.ArrayList;
import java.util.Arrays;
import java.util.HashMap;
import java.util.HashSet;
import java.util.List;
Expand Down Expand Up @@ -241,74 +242,23 @@ private static Node readNode(String nodeId, JsonParser parser, Scheme scheme) th
}

Map<String, List<String>> realAttributes = new HashMap<>(protoAttributes.size());
List<String> keys = new ArrayList<>(protoAttributes.keySet());
for (String key : keys) {
if (key.endsWith(".0")) {
String realKey = key.substring(0, key.length() - 2);
List<String> values = new ArrayList<>();
int i = 0;
while (true) {
String value = protoAttributes.remove(realKey + "." + i);
if (value == null) {
break;
}
values.add(value);
i++;
}
realAttributes.put(realKey, unmodifiableList(values));
}
}
for (Map.Entry<String, String> entry : protoAttributes.entrySet()) {
realAttributes.put(entry.getKey(), singletonList(entry.getValue()));
}

if (version.startsWith("2.")) {
/*
* 2.x doesn't send roles, instead we try to read them from
* attributes.
*/
boolean clientAttribute = v2RoleAttributeValue(realAttributes, "client", false);
Boolean masterAttribute = v2RoleAttributeValue(realAttributes, "master", null);
Boolean dataAttribute = v2RoleAttributeValue(realAttributes, "data", null);
if ((masterAttribute == null && false == clientAttribute) || masterAttribute) {
roles.add("master");
if (entry.getValue().startsWith("[")) {
// Convert string array to list
String value = entry.getValue();
String[] values = value.substring(1, value.length() - 1).split(", ");
realAttributes.put(entry.getKey(), unmodifiableList(Arrays.asList(values)));
} else {
realAttributes.put(entry.getKey(), singletonList(entry.getValue()));
}
if ((dataAttribute == null && false == clientAttribute) || dataAttribute) {
roles.add("data");
}
} else {
assert sawRoles : "didn't see roles for [" + nodeId + "]";
}

assert sawRoles : "didn't see roles for [" + nodeId + "]";
assert boundHosts.contains(publishedHost) : "[" + nodeId + "] doesn't make sense! publishedHost should be in boundHosts";
logger.trace("adding node [" + nodeId + "]");
return new Node(publishedHost, boundHosts, name, version, new Roles(roles), unmodifiableMap(realAttributes));
}

/**
* Returns {@code defaultValue} if the attribute didn't come back,
* {@code true} or {@code false} if it did come back as
* either of those, or throws an IOException if the attribute
* came back in a strange way.
*/
private static Boolean v2RoleAttributeValue(Map<String, List<String>> attributes, String name, Boolean defaultValue)
throws IOException {
List<String> valueList = attributes.remove(name);
if (valueList == null) {
return defaultValue;
}
if (valueList.size() != 1) {
throw new IOException("expected only a single attribute value for [" + name + "] but got " + valueList);
}
switch (valueList.get(0)) {
case "true":
return true;
case "false":
return false;
default:
throw new IOException("expected [" + name + "] to be either [true] or [false] but was [" + valueList.get(0) + "]");
}
}

/**
* The supported host schemes.
*/
Expand Down
92 changes: 27 additions & 65 deletions ...t/sniffer/src/test/java/org/opensearch/client/sniff/OpenSearchNodesSnifferParseTests.java
View file
Original file line number Diff line number Diff line change
Expand Up @@ -45,8 +45,8 @@
import java.io.IOException;
import java.io.InputStream;
import java.util.Arrays;
import java.util.HashMap;
import java.util.HashSet;
import java.util.LinkedHashMap;
import java.util.List;
import java.util.Map;
import java.util.Set;
Expand Down Expand Up @@ -85,59 +85,31 @@ private void checkFile(String file, Node... expected) throws IOException {
}
}

public void test2x() throws IOException {
checkFile(
"2.0.0_nodes_http.json",
node(9200, "m1", "2.0.0", true, false, false),
node(9201, "m2", "2.0.0", true, true, false),
node(9202, "m3", "2.0.0", true, false, false),
node(9203, "d1", "2.0.0", false, true, false),
node(9204, "d2", "2.0.0", false, true, false),
node(9205, "d3", "2.0.0", false, true, false),
node(9206, "c1", "2.0.0", false, false, false),
node(9207, "c2", "2.0.0", false, false, false)
);
}

public void test5x() throws IOException {
checkFile(
"5.0.0_nodes_http.json",
node(9200, "m1", "5.0.0", true, false, true),
node(9201, "m2", "5.0.0", true, true, true),
node(9202, "m3", "5.0.0", true, false, true),
node(9203, "d1", "5.0.0", false, true, true),
node(9204, "d2", "5.0.0", false, true, true),
node(9205, "d3", "5.0.0", false, true, true),
node(9206, "c1", "5.0.0", false, false, true),
node(9207, "c2", "5.0.0", false, false, true)
);
}

public void test6x() throws IOException {
public void test1x() throws IOException {
checkFile(
"6.0.0_nodes_http.json",
node(9200, "m1", "6.0.0", true, false, true),
node(9201, "m2", "6.0.0", true, true, true),
node(9202, "m3", "6.0.0", true, false, true),
node(9203, "d1", "6.0.0", false, true, true),
node(9204, "d2", "6.0.0", false, true, true),
node(9205, "d3", "6.0.0", false, true, true),
node(9206, "c1", "6.0.0", false, false, true),
node(9207, "c2", "6.0.0", false, false, true)
"1.0.0_nodes_http.json",
node(9200, "m1", "1.0.0", "master", "ingest"),
node(9201, "m2", "1.0.0", "master", "data", "ingest"),
node(9202, "m3", "1.0.0", "master", "ingest"),
node(9203, "d1", "1.0.0", "data", "ingest"),
node(9204, "d2", "1.0.0", "data", "ingest"),
node(9205, "d3", "1.0.0", "data", "ingest"),
node(9206, "c1", "1.0.0", "ingest"),
node(9207, "c2", "1.0.0", "ingest")
);
}

public void test7x() throws IOException {
public void test2x() throws IOException {
checkFile(
"7.3.0_nodes_http.json",
node(9200, "m1", "7.3.0", "master", "ingest"),
node(9201, "m2", "7.3.0", "master", "data", "ingest"),
node(9202, "m3", "7.3.0", "master", "ingest"),
node(9203, "d1", "7.3.0", "data", "ingest", "ml"),
node(9204, "d2", "7.3.0", "data", "ingest"),
node(9205, "d3", "7.3.0", "data", "ingest"),
node(9206, "c1", "7.3.0", "ingest"),
node(9207, "c2", "7.3.0", "ingest")
"2.0.0_nodes_http.json",
node(9200, "m1", "2.0.0", "cluster_manager", "ingest"),
node(9201, "m2", "2.0.0", "cluster_manager", "data", "ingest"),
node(9202, "m3", "2.0.0", "cluster_manager", "ingest"),
node(9203, "d1", "2.0.0", "data", "ingest"),
node(9204, "d2", "2.0.0", "data", "ingest"),
node(9205, "d3", "2.0.0", "data", "ingest"),
node(9206, "c1", "2.0.0", "ingest"),
node(9207, "c2", "2.0.0", "ingest")
);
}

Expand All @@ -163,32 +135,22 @@ public void testParsingPublishAddressWithES7Format() throws IOException {
assertEquals("http", nodes.get(0).getHost().getSchemeName());
}

private Node node(int port, String name, String version, boolean master, boolean data, boolean ingest) {
final Set<String> roles = new TreeSet<>();
if (master) {
roles.add("master");
}
if (data) {
roles.add("data");
}
if (ingest) {
roles.add("ingest");
}
return node(port, name, version, roles);
}

private Node node(int port, String name, String version, String... roles) {
return node(port, name, version, new TreeSet<>(Arrays.asList(roles)));
}

private Node node(int port, String name, String version, Set<String> roles) {
HttpHost host = new HttpHost("127.0.0.1", port);
Set<HttpHost> boundHosts = new HashSet<>(2);
boundHosts.add(host);
boundHosts.add(new HttpHost("[::1]", port));
Map<String, List<String>> attributes = new HashMap<>();
boundHosts.add(host);
Map<String, List<String>> attributes = new LinkedHashMap<>(); // LinkedHashMap to preserve insertion order
attributes.put("dummy", singletonList("everyone_has_me"));
attributes.put("number", singletonList(name.substring(1)));
if (!version.startsWith("1.0") && !version.startsWith("1.1")) {
// Shard Indexing Pressure feature is added in version 1.2.0
attributes.put("shard_indexing_pressure_enabled", singletonList(Boolean.TRUE.toString()));
}
attributes.put("array", Arrays.asList(name.substring(0, 1), name.substring(1)));
return new Node(host, boundHosts, name, version, new Roles(new TreeSet<>(roles)), attributes);
}
Expand Down
16 changes: 5 additions & 11 deletions client/sniffer/src/test/java/org/opensearch/client/sniff/OpenSearchNodesSnifferTests.java
View file
Original file line number Diff line number Diff line change
Expand Up @@ -234,7 +234,7 @@ private static SniffResponse buildSniffResponse(OpenSearchNodesSniffer.Scheme sc

final Set<String> nodeRoles = new TreeSet<>();
if (randomBoolean()) {
nodeRoles.add("master");
nodeRoles.add("cluster_manager");
}
if (randomBoolean()) {
nodeRoles.add("data");
Expand Down Expand Up @@ -283,12 +283,12 @@ private static SniffResponse buildSniffResponse(OpenSearchNodesSniffer.Scheme sc
generator.writeEndObject();
}

List<String> roles = Arrays.asList(new String[] { "master", "data", "ingest" });
List<String> roles = Arrays.asList(new String[] { "cluster_manager", "data", "ingest" });
Collections.shuffle(roles, getRandom());
generator.writeArrayFieldStart("roles");
for (String role : roles) {
if ("master".equals(role) && node.getRoles().isMasterEligible()) {
generator.writeString("master");
if ("cluster_manager".equals(role) && node.getRoles().isMasterEligible()) {
generator.writeString("cluster_manager");
}
if ("data".equals(role) && node.getRoles().isData()) {
generator.writeString("data");
Expand All @@ -307,13 +307,7 @@ private static SniffResponse buildSniffResponse(OpenSearchNodesSniffer.Scheme sc
if (numAttributes > 0) {
generator.writeObjectFieldStart("attributes");
for (Map.Entry<String, List<String>> entry : attributes.entrySet()) {
if (entry.getValue().size() == 1) {
generator.writeStringField(entry.getKey(), entry.getValue().get(0));
} else {
for (int v = 0; v < entry.getValue().size(); v++) {
generator.writeStringField(entry.getKey() + "." + v, entry.getValue().get(v));
}
}
generator.writeStringField(entry.getKey(), entry.getValue().toString());
}
generator.writeEndObject();
}
Expand Down
Loading

0 comments on commit ab478ba

Please sign in to comment.