Skip to content

[elasticsearch,elasticsearch5] Add Elasticsearch 5.x binding #925

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.

Sign up for GitHub

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

Merged
jasontedor merged 27 commits into from
Aug 10, 2017
Merged

[elasticsearch,elasticsearch5] Add Elasticsearch 5.x binding #925

Show file tree
Hide file tree
Changes from 1 commit
Commits
Show all changes
27 commits
Select commit Hold shift + click to select a range
c3cfc98
[elasticsearch,elasticsearch5] Add Elasticsearch 5.x binding
risdenk Feb 7, 2017
c52c438
Start REST integration tests
risdenk Feb 8, 2017
4c84ffa
Merge branch 'master' into initial-es5
jasontedor Aug 7, 2017
b1e1d48
Elasticsearch 5: Set dependency to version 5.5.1
jasontedor Aug 7, 2017
a3ba64d
Elasticsearch 5: Avoid adding path.home if not set
jasontedor Aug 7, 2017
23163cc
Elasticsearch 5: Fix handling of settings
jasontedor Aug 7, 2017
d66e856
Elasticsearch 5: Remove support for embedded node
jasontedor Aug 7, 2017
e47e90e
Elasticsearch 5: Use auto-IDs and implements scan
jasontedor Aug 8, 2017
db8674a
Elasticsearch 5: Remove path.home setting
jasontedor Aug 8, 2017
0eb22d1
Elasticsearch 5: Fix unreleased bugs in client
jasontedor Aug 8, 2017
bc69e7e
Elasticsearch 5: Complete REST implementation
jasontedor Aug 8, 2017
f1eed61
Elasticsearch 5: Code cleanup
jasontedor Aug 8, 2017
42bb148
Elasticsearch 5: Fix issues for tests to pass
jasontedor Aug 9, 2017
382bdc7
Elasticsearch 5: Update docs
jasontedor Aug 9, 2017
6d14ff4
Elasticsearch 5: Only activate plugin on JDK 8
jasontedor Aug 9, 2017
1774d62
Elasticsearch 5: Format transport client logs
jasontedor Aug 9, 2017
6ae7050
Elasticsearch 5: Revert unrelated changes
jasontedor Aug 10, 2017
a65dcb7
Elasticsearch 5: Log transport client on stderr
jasontedor Aug 10, 2017
e30441c
Elasticsearch 5: Remove unneeded output statements
jasontedor Aug 10, 2017
d791ae6
Elasticsearch 5: Change es.newdb to es.new_index
jasontedor Aug 10, 2017
9267038
Elasticsearch 5: Close content builder
jasontedor Aug 10, 2017
2250903
Elasticsearch 5: Remove core POM change
jasontedor Aug 10, 2017
5d840ef
Elasticsearch 5: Fix Javadocs in REST client tests
jasontedor Aug 10, 2017
b9f8539
Elasticsearch 5: Collapse tests to one class
jasontedor Aug 10, 2017
38d45b0
Elasticsearch 5: Mark constant as final
jasontedor Aug 10, 2017
082dc30
Elasticsearch 5: Use provided constants for status codes
jasontedor Aug 10, 2017
033ff88
Elasticsearch 5: Remove extraneous newline
jasontedor Aug 10, 2017
File filter

Filter by extension

Filter by extension
Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
Prev Previous commit
Next Next commit
Elasticsearch 5: Code cleanup 
This commit is a straightforward code cleanup of the Elasticsearch 5
transport client and REST client implementations.
  • Loading branch information
@jasontedor
jasontedor committed Aug 8, 2017
commit f1eed61e731173a55802ba379d8987acb11b19dd
35 changes: 35 additions & 0 deletions elasticsearch5/src/main/java/com/yahoo/ycsb/db/elasticsearch5/Elasticsearch5.java
View file
Open in desktop
Original file line number Diff line number Diff line change
@@ -0,0 +1,35 @@
/*
* Copyright (c) 2017 YCSB contributors. All rights reserved.
* <p>
* Licensed under the Apache License, Version 2.0 (the "License"); you
* may not use this file except in compliance with the License. You
* may obtain a copy of the License at
* <p>
* http://www.apache.org/licenses/LICENSE-2.0
* <p>
* Unless required by applicable law or agreed to in writing, software
* distributed under the License is distributed on an "AS IS" BASIS,
* WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or
* implied. See the License for the specific language governing
* permissions and limitations under the License. See accompanying
* LICENSE file.
*/

package com.yahoo.ycsb.db.elasticsearch5;

import java.util.Properties;

final class Elasticsearch5 {

private Elasticsearch5() {

}

static final String KEY = "key";

static int parseIntegerProperty(final Properties properties, final String key, final int defaultValue) {
final String value = properties.getProperty(key);
return value == null ? defaultValue : Integer.parseInt(value);
}

}
61 changes: 29 additions & 32 deletions elasticsearch5/src/main/java/com/yahoo/ycsb/db/elasticsearch5/ElasticsearchClient.java
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -26,7 +26,6 @@
import org.elasticsearch.action.admin.indices.create.CreateIndexRequest;
import org.elasticsearch.action.delete.DeleteResponse;
import org.elasticsearch.action.search.SearchResponse;
import org.elasticsearch.client.Client;
import org.elasticsearch.client.Requests;
import org.elasticsearch.client.transport.TransportClient;
import org.elasticsearch.common.settings.Settings;
Expand All @@ -48,6 +47,8 @@
import java.util.Set;
import java.util.Vector;

import static com.yahoo.ycsb.db.elasticsearch5.Elasticsearch5.KEY;
import static com.yahoo.ycsb.db.elasticsearch5.Elasticsearch5.parseIntegerProperty;
import static org.elasticsearch.common.settings.Settings.Builder;
import static org.elasticsearch.common.xcontent.XContentFactory.jsonBuilder;

Expand All @@ -61,7 +62,7 @@ public class ElasticsearchClient extends DB {
private static final String DEFAULT_REMOTE_HOST = "localhost:9300";
private static final int NUMBER_OF_SHARDS = 1;
private static final int NUMBER_OF_REPLICAS = 0;
private Client client;
private TransportClient client;
private String indexKey;

/**
Expand All @@ -75,11 +76,11 @@ public void init() throws DBException {

this.indexKey = props.getProperty("es.index.key", DEFAULT_INDEX_KEY);

int numberOfShards = parseIntegerProperty(props, "es.number_of_shards", NUMBER_OF_SHARDS);
int numberOfReplicas = parseIntegerProperty(props, "es.number_of_replicas", NUMBER_OF_REPLICAS);
final int numberOfShards = parseIntegerProperty(props, "es.number_of_shards", NUMBER_OF_SHARDS);
final int numberOfReplicas = parseIntegerProperty(props, "es.number_of_replicas", NUMBER_OF_REPLICAS);

Boolean newdb = Boolean.parseBoolean(props.getProperty("es.newdb", "false"));
Builder settings = Settings.builder().put("cluster.name", DEFAULT_CLUSTER_NAME);
final Boolean newdb = Boolean.parseBoolean(props.getProperty("es.newdb", "false"));
final Builder settings = Settings.builder().put("cluster.name", DEFAULT_CLUSTER_NAME);

// if properties file contains elasticsearch user defined properties
// add it to the settings file (will overwrite the defaults).
Expand All @@ -92,30 +93,33 @@ public void init() throws DBException {
}
}
final String clusterName = settings.get("cluster.name");
System.err.println("Elasticsearch starting node = " + clusterName);
System.out.println("Elasticsearch cluster name = " + clusterName);

@risdenk risdenk Aug 10, 2017

Copy link
Copy Markdown
Collaborator Author

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

Think this should go to System.err if we need it. Metrics should go to standard out.

@jasontedor jasontedor Aug 10, 2017

Copy link
Copy Markdown
Collaborator

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

I dropped it: e30441c


settings.put("client.transport.sniff", true)
.put("client.transport.ignore_cluster_name", false)
.put("client.transport.ping_timeout", "30s")
.put("client.transport.nodes_sampler_interval", "30s");
// Default it to localhost:9300
String[] nodeList = props.getProperty("es.hosts.list", DEFAULT_REMOTE_HOST).split(",");
final String[] nodeList = props.getProperty("es.hosts.list", DEFAULT_REMOTE_HOST).split(",");
System.out.println("Elasticsearch Remote Hosts = " + props.getProperty("es.hosts.list", DEFAULT_REMOTE_HOST));

@risdenk risdenk Aug 10, 2017

Copy link
Copy Markdown
Collaborator Author

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

Think this should go to System.err if we need it. Metrics should go to standard out.

@jasontedor jasontedor Aug 10, 2017

Copy link
Copy Markdown
Collaborator

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

I dropped it: e30441c

TransportClient tClient = new PreBuiltTransportClient(settings.build());
client = new PreBuiltTransportClient(settings.build());
for (String h : nodeList) {
String[] nodes = h.split(":");

final InetAddress address;
try {
tClient.addTransportAddress(new InetSocketTransportAddress(
InetAddress.getByName(nodes[0]),
Integer.parseInt(nodes[1])
));
} catch (NumberFormatException e) {
throw new IllegalArgumentException("Unable to parse port number.", e);
address = InetAddress.getByName(nodes[0]);
} catch (UnknownHostException e) {
throw new IllegalArgumentException("Unable to Identify host.", e);
throw new IllegalArgumentException("unable to identity host [" + nodes[0]+ "]", e);
}
final int port;
try {
port = Integer.parseInt(nodes[1]);
} catch (final NumberFormatException e) {
throw new IllegalArgumentException("unable to parse port [" + nodes[1] + "]", e);
}
client.addTransportAddress(new InetSocketTransportAddress(address, port));
}
client = tClient;

final boolean exists =
client.admin().indices()
Expand All @@ -136,11 +140,6 @@ public void init() throws DBException {
client.admin().cluster().health(new ClusterHealthRequest().waitForGreenStatus()).actionGet();
}

private int parseIntegerProperty(final Properties properties, final String key, final int defaultValue) {
final String value = properties.getProperty(key);
return value == null ? defaultValue : Integer.parseInt(value);
}

@Override
public void cleanup() throws DBException {
if (client != null) {
Expand All @@ -158,8 +157,7 @@ public Status insert(String table, String key, Map<String, ByteIterator> values)
for (final Entry<String, String> entry : StringByteIterator.getStringMap(values).entrySet()) {
doc.field(entry.getKey(), entry.getValue());
}

doc.field("key", key);
doc.field(KEY, key);
doc.endObject();

client.prepareIndex(indexKey, table).setSource(doc).execute().actionGet();
Expand All @@ -184,9 +182,9 @@ public Status delete(final String table, final String key) {
final DeleteResponse deleteResponse = client.prepareDelete(indexKey, table, id).execute().actionGet();
if (deleteResponse.status().equals(RestStatus.NOT_FOUND)) {
return Status.NOT_FOUND;
} else {
return Status.OK;
}

return Status.OK;
} catch (final Exception e) {
e.printStackTrace();

@risdenk risdenk Aug 10, 2017

Copy link
Copy Markdown
Collaborator Author

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

I think this should go to stderr. Not sure it does by default.

@jasontedor jasontedor Aug 10, 2017

Copy link
Copy Markdown
Collaborator

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

It does, Exception#printStackTrace goes to stderr by default.

return Status.ERROR;
Expand All @@ -213,14 +211,14 @@ public Status read(
}
} else {
for (final Map.Entry<String, SearchHitField> e : hit.getFields().entrySet()) {
if ("key".equals(e.getKey())) {
if (KEY.equals(e.getKey())) {
continue;
}
result.put(e.getKey(), new StringByteIterator((String) e.getValue().getValue()));
}
}
return Status.OK;

return Status.OK;
} catch (final Exception e) {
e.printStackTrace();

@risdenk risdenk Aug 10, 2017

Copy link
Copy Markdown
Collaborator Author

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

should go to stderr

@jasontedor jasontedor Aug 10, 2017

Copy link
Copy Markdown
Collaborator

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

Exception#printStackTrace goes to stderr by default.

return Status.ERROR;
Expand All @@ -243,7 +241,6 @@ public Status update(final String table, final String key, final Map<String, Byt
client.prepareIndex(indexKey, table, hit.getId()).setSource(hit.getSource()).get();

return Status.OK;

} catch (final Exception e) {
e.printStackTrace();

@risdenk risdenk Aug 10, 2017

Copy link
Copy Markdown
Collaborator Author

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

stderr

@jasontedor jasontedor Aug 10, 2017

Copy link
Copy Markdown
Collaborator

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

Exception#printStackTrace goes to stderr by default.

return Status.ERROR;
Expand All @@ -258,7 +255,7 @@ public Status scan(
final Set<String> fields,
final Vector<HashMap<String, ByteIterator>> result) {
try {
final RangeQueryBuilder query = new RangeQueryBuilder("key").gte(startkey);
final RangeQueryBuilder query = new RangeQueryBuilder(KEY).gte(startkey);
final SearchResponse response = client.prepareSearch(indexKey).setQuery(query).setSize(recordcount).get();

for (final SearchHit hit : response.getHits()) {
Expand All @@ -271,7 +268,7 @@ public Status scan(
} else {
entry = new HashMap<>(hit.getFields().size());
for (final Map.Entry<String, SearchHitField> field : hit.getFields().entrySet()) {
if ("key".equals(field.getKey())) {
if (KEY.equals(field.getKey())) {
continue;
}
entry.put(field.getKey(), new StringByteIterator((String) field.getValue().getValue()));
Expand All @@ -288,7 +285,7 @@ public Status scan(


private SearchResponse search(final String table, final String key) {
return client.prepareSearch(indexKey).setTypes(table).setQuery(new TermQueryBuilder("key", key)).get();
return client.prepareSearch(indexKey).setTypes(table).setQuery(new TermQueryBuilder(KEY, key)).get();
}

}
68 changes: 28 additions & 40 deletions elasticsearch5/src/main/java/com/yahoo/ycsb/db/elasticsearch5/ElasticsearchRestClient.java
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -32,7 +32,6 @@
import org.codehaus.jackson.map.ObjectMapper;
import org.elasticsearch.client.Response;
import org.elasticsearch.client.RestClient;
import org.elasticsearch.common.settings.Settings;
import org.elasticsearch.common.xcontent.XContentBuilder;

import java.io.IOException;
Expand All @@ -46,8 +45,9 @@
import java.util.Set;
import java.util.Vector;

import static com.yahoo.ycsb.db.elasticsearch5.Elasticsearch5.KEY;
import static com.yahoo.ycsb.db.elasticsearch5.Elasticsearch5.parseIntegerProperty;
import static java.util.Collections.emptyMap;
import static org.elasticsearch.common.settings.Settings.Builder;
import static org.elasticsearch.common.xcontent.XContentFactory.jsonBuilder;

/**
Expand All @@ -74,31 +74,15 @@ public void init() throws DBException {

this.indexKey = props.getProperty("es.index.key", DEFAULT_INDEX_KEY);

int numberOfShards = Integer.valueOf(props.getProperty("es.number_of_shards",
String.valueOf(NUMBER_OF_SHARDS)));
int numberOfReplicas = Integer.valueOf(props.getProperty("es.number_of_replicas",
String.valueOf(NUMBER_OF_REPLICAS)));

Boolean newdb = Boolean.parseBoolean(props.getProperty("es.newdb", "false"));
Builder settings = Settings.builder().put("cluster.name", DEFAULT_CLUSTER_NAME);

// if properties file contains elasticsearch user defined properties
// add it to the settings file (will overwrite the defaults).
for (final Map.Entry<Object, Object> e : props.entrySet()) {
if (e.getKey() instanceof String) {
final String key = (String) e.getKey();
if (key.startsWith("es.setting.")) {
settings.put(key.substring("es.setting.".length()), e.getValue());
}
}
}
final String clusterName = settings.get("cluster.name");
System.err.println("Elasticsearch starting node = " + clusterName);
final int numberOfShards = parseIntegerProperty(props, "es.number_of_shards", NUMBER_OF_SHARDS);
final int numberOfReplicas = parseIntegerProperty(props, "es.number_of_replicas", NUMBER_OF_REPLICAS);

final Boolean newdb = Boolean.parseBoolean(props.getProperty("es.newdb", "false"));

String[] nodeList = props.getProperty("es.hosts.list", DEFAULT_REMOTE_HOST).split(",");
final String[] nodeList = props.getProperty("es.hosts.list", DEFAULT_REMOTE_HOST).split(",");
System.out.println("Elasticsearch Remote Hosts = " + props.getProperty("es.hosts.list", DEFAULT_REMOTE_HOST));

@risdenk risdenk Aug 10, 2017

Copy link
Copy Markdown
Collaborator Author

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

should go to stderr

@jasontedor jasontedor Aug 10, 2017 •
edited
Loading

Copy link
Copy Markdown
Collaborator

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

I removed this logging statement, it's superfluous anyway: e30441c


List<HttpHost> esHttpHosts = new ArrayList<>(nodeList.length);
final List<HttpHost> esHttpHosts = new ArrayList<>(nodeList.length);
for (String h : nodeList) {
String[] nodes = h.split(":");
esHttpHosts.add(new HttpHost(nodes[0], Integer.valueOf(nodes[1]), "http"));
Expand Down Expand Up @@ -204,7 +188,7 @@ public void cleanup() throws DBException {
public Status insert(final String table, final String key, final Map<String, ByteIterator> values) {
try {
final Map<String, String> data = StringByteIterator.getStringMap(values);
data.put("key", key);
data.put(KEY, key);

final Response response = restClient.performRequest(
"POST",
Expand All @@ -224,7 +208,7 @@ public Status insert(final String table, final String key, final Map<String, Byt
}

@Override
public Status delete(String table, String key) {
public Status delete(final String table, final String key) {
try {
final Response searchResponse = search(table, key);
final int statusCode = searchResponse.getStatusLine().getStatusCode();
Expand All @@ -240,7 +224,8 @@ public Status delete(String table, String key) {
if (total == 0) {
return Status.NOT_FOUND;
}
@SuppressWarnings("unchecked") final Map<String, Object> hit = (Map<String, Object>)((List<Object>)hits.get("hits")).get(0);
@SuppressWarnings("unchecked") final Map<String, Object> hit =
(Map<String, Object>)((List<Object>)hits.get("hits")).get(0);
@SuppressWarnings("unchecked") final Map<String, Object> source = (Map<String, Object>)hit.get("_source");
final Response deleteResponse =
restClient.performRequest("DELETE", "/" + indexKey + "/" + table + "/" + source.get("_id"));
Expand Down Expand Up @@ -276,15 +261,16 @@ public Status read(
if (total == 0) {
return Status.NOT_FOUND;
}
@SuppressWarnings("unchecked") final Map<String, Object> hit = (Map<String, Object>)((List<Object>)hits.get("hits")).get(0);
@SuppressWarnings("unchecked") final Map<String, Object> hit =
(Map<String, Object>)((List<Object>)hits.get("hits")).get(0);
@SuppressWarnings("unchecked") final Map<String, Object> source = (Map<String, Object>)hit.get("_source");
if (fields != null) {
for (final String field : fields) {
result.put(field, new StringByteIterator((String) source.get(field)));
}
} else {
for (final Map.Entry<String, Object> e : source.entrySet()) {
if ("key".equals(e.getKey())) {
if (KEY.equals(e.getKey())) {
continue;
}
result.put(e.getKey(), new StringByteIterator((String) e.getValue()));
Expand All @@ -299,7 +285,7 @@ public Status read(
}

@Override
public Status update(String table, String key, Map<String, ByteIterator> values) {
public Status update(final String table, final String key, final Map<String, ByteIterator> values) {
try {
final Response searchResponse = search(table, key);
final int statusCode = searchResponse.getStatusLine().getStatusCode();
Expand All @@ -315,7 +301,8 @@ public Status update(String table, String key, Map<String, ByteIterator> values)
if (total == 0) {
return Status.NOT_FOUND;
}
@SuppressWarnings("unchecked") final Map<String, Object> hit = (Map<String, Object>) ((List<Object>) hits.get("hits")).get(0);
@SuppressWarnings("unchecked") final Map<String, Object> hit =
(Map<String, Object>) ((List<Object>) hits.get("hits")).get(0);
@SuppressWarnings("unchecked") final Map<String, Object> source = (Map<String, Object>) hit.get("_source");
for (final Map.Entry<String, String> entry : StringByteIterator.getStringMap(values).entrySet()) {
source.put(entry.getKey(), entry.getValue());
Expand All @@ -338,18 +325,18 @@ public Status update(String table, String key, Map<String, ByteIterator> values)

@Override
public Status scan(
String table,
String startkey,
int recordcount,
Set<String> fields,
Vector<HashMap<String, ByteIterator>> result) {
final String table,
final String startkey,
final int recordcount,
final Set<String> fields,
final Vector<HashMap<String, ByteIterator>> result) {
try {
final Response response;
try (XContentBuilder builder = jsonBuilder()) {
builder.startObject();
builder.startObject("query");
builder.startObject("range");
builder.startObject("key");
builder.startObject(KEY);
builder.field("gte", startkey);
builder.endObject();
builder.endObject();
Expand All @@ -359,7 +346,8 @@ public Status scan(
response = search(table, builder);
@SuppressWarnings("unchecked") final Map<String, Object> map = map(response);
@SuppressWarnings("unchecked") final Map<String, Object> hits = (Map<String, Object>)map.get("hits");
@SuppressWarnings("unchecked") final List<Map<String, Object>> list = (List<Map<String, Object>>) hits.get("hits");
@SuppressWarnings("unchecked") final List<Map<String, Object>> list =
(List<Map<String, Object>>) hits.get("hits");

for (final Map<String, Object> hit : list) {
@SuppressWarnings("unchecked") final Map<String, Object> source = (Map<String, Object>)hit.get("_source");
Expand All @@ -372,7 +360,7 @@ public Status scan(
} else {
entry = new HashMap<>(hit.size());
for (final Map.Entry<String, Object> field : source.entrySet()) {
if ("key".equals(field.getKey())) {
if (KEY.equals(field.getKey())) {
continue;
}
entry.put(field.getKey(), new StringByteIterator((String) field.getValue()));
Expand All @@ -393,7 +381,7 @@ private Response search(final String table, final String key) throws IOException
builder.startObject();
builder.startObject("query");
builder.startObject("term");
builder.field("key", key);
builder.field(KEY, key);
builder.endObject();
builder.endObject();
builder.endObject();
Expand Down