joerg
/
elx
1
0
Fork 0
Code Issues Pull Requests Packages Projects Releases Wiki Activity

re-enable bulk logging, parameter cleanup, leaner API, strict IndexDefinition based

Browse Source
2.2.1.32
Jörg Prante 3 years ago
parent a9743dedf7
commit 719e9ee05f
42 changed files with 536 additions and 796 deletions
Show Stats Download Patch File Download Diff File

67
elx-api/src/main/java/org/xbib/elx/api/AdminClient.java
Unescape Escape View File

@ -1,30 +1,17 @@
package org.xbib.elx.api;
import org.elasticsearch.common.settings.Settings;
import java.io.IOException;
import java.util.List;
import java.util.Map;
import java.util.concurrent.TimeUnit;
/**
* Interface for extended managing and indexing methods of an Elasticsearch client.
*/
public interface AdminClient extends BasicClient {
/**
* Build index definition from settings.
*
* @param index the index name
* @param settings the settings for the index
* @return index definition
* @throws IOException if settings/mapping URL is invalid/malformed
*/
IndexDefinition buildIndexDefinitionFromSettings(String index, Settings settings) throws IOException;
Map<String, ?> getMapping(String index) throws IOException;
Map<String, ?> getMapping(IndexDefinition indexDefinition) throws IOException;
void checkMapping(String index);
void checkMapping(IndexDefinition indexDefinition);
/**
* Delete an index.
@ -33,15 +20,6 @@ public interface AdminClient extends BasicClient {
*/
AdminClient deleteIndex(IndexDefinition indexDefinition);
/**
* Delete an index.
*
* @param index index
* @return this
*/
AdminClient deleteIndex(String index);
/**
* Update replica level.
* @param indexDefinition the index definition
@ -51,18 +29,6 @@ public interface AdminClient extends BasicClient {
*/
AdminClient updateReplicaLevel(IndexDefinition indexDefinition, int level) throws IOException;
/**
* Update replica level.
*
* @param index index
* @param level the replica level
* @param maxWaitTime maximum wait time
* @param timeUnit time unit
* @return this
* @throws IOException if replica setting could not be updated
*/
AdminClient updateReplicaLevel(String index, int level, long maxWaitTime, TimeUnit timeUnit) throws IOException;
/**
* Get replica level.
* @param indexDefinition the index name
@ -70,13 +36,6 @@ public interface AdminClient extends BasicClient {
*/
int getReplicaLevel(IndexDefinition indexDefinition);
/**
* Get replica level.
* @param index the index name
* @return the replica level of the index
*/
int getReplicaLevel(String index);
/**
* Force segment merge of an index.
* @param indexDefinition the index definition
@ -84,15 +43,6 @@ public interface AdminClient extends BasicClient {
*/
boolean forceMerge(IndexDefinition indexDefinition);
/**
* Force segment merge of an index.
* @param index the index
* @param maxWaitTime maximum wait time
* @param timeUnit time unit
* @return this
*/
boolean forceMerge(String index, long maxWaitTime, TimeUnit timeUnit);
/**
* Resolve alias.
*
@ -117,15 +67,6 @@ public interface AdminClient extends BasicClient {
*/
Map<String, String> getAliases(String index);
/**
* Shift from one index to another.
* @param indexDefinition the index definition
* @param additionalAliases new aliases
* @return this
*/
IndexShiftResult shiftIndex(IndexDefinition indexDefinition,
List<String> additionalAliases);
/**
* Shift from one index to another.
* @param indexDefinition the index definition
@ -148,10 +89,10 @@ public interface AdminClient extends BasicClient {
/**
* Find the timestamp of the most recently indexed document in the index.
*
* @param index the index name
* @param indexDefinition the index
* @param timestampfieldname the timestamp field name
* @return millis UTC millis of the most recent document
* @throws IOException if most rcent document can not be found
*/
Long mostRecentDocument(String index, String timestampfieldname) throws IOException;
Long mostRecentDocument(IndexDefinition indexDefinition, String timestampfieldname) throws IOException;
}

4
elx-api/src/main/java/org/xbib/elx/api/BasicClient.java
Unescape Escape View File

@ -56,7 +56,7 @@ public interface BasicClient extends Closeable {
void waitForShards(long maxWaitTime, TimeUnit timeUnit);
long getSearchableDocs(String index);
long getSearchableDocs(IndexDefinition index);
boolean isIndexExists(String index);
boolean isIndexExists(IndexDefinition index);
}

84
elx-api/src/main/java/org/xbib/elx/api/BulkClient.java
Unescape Escape View File

@ -23,31 +23,44 @@ public interface BulkClient extends BasicClient, Flushable {
*/
void newIndex(IndexDefinition indexDefinition) throws IOException;
/**
* Start bulk mode for indexes.
* @param indexDefinition index definition
* @throws IOException if bulk could not be started
*/
void startBulk(IndexDefinition indexDefinition) throws IOException;
/**
* Stop bulk mode.
*
* @param indexDefinition index definition
* @throws IOException if bulk could not be startet
*/
void stopBulk(IndexDefinition indexDefinition) throws IOException;
/**
* Add index request. Each request will be added to a queue for bulking requests.
* Submitting request will be done when limits are exceeded.
*
* @param index the index
* @param type the type
* @param indexDefinition the index definition
* @param id the id
* @param create true if document must be created
* @param source the source
* @return this
*/
BulkClient index(String index, String type, String id, boolean create, BytesReference source);
BulkClient index(IndexDefinition indexDefinition, String id, boolean create, BytesReference source);
/**
* Index request. Each request will be added to a queue for bulking requests.
* Submitting request will be done when limits are exceeded.
*
* @param index the index
* @param type the type
* @param indexDefinition the index definition
* @param id the id
* @param create true if document is to be created, false otherwise
* @param source the source
* @return this client methods
*/
BulkClient index(String index, String type, String id, boolean create, String source);
BulkClient index(IndexDefinition indexDefinition, String id, boolean create, String source);
/**
* Index request. Each request will be added to a queue for bulking requests.
@ -61,12 +74,11 @@ public interface BulkClient extends BasicClient, Flushable {
/**
* Delete request.
*
* @param index the index
* @param type the type
* @param indexDefinition the index definition
* @param id the id
* @return this
*/
BulkClient delete(String index, String type, String id);
BulkClient delete(IndexDefinition indexDefinition, String id);
/**
* Delete request. Each request will be added to a queue for bulking requests.
@ -82,24 +94,22 @@ public interface BulkClient extends BasicClient, Flushable {
* Submitting request will be done when bulk limits are exceeded.
* Note that updates only work correctly when all operations between nodes are synchronized.
*
* @param index the index
* @param type the type
* @param indexDefinition the index definition
* @param id the id
* @param source the source
* @return this
*/
BulkClient update(String index, String type, String id, BytesReference source);
BulkClient update(IndexDefinition indexDefinition, String id, BytesReference source);
/**
* Update document. Use with precaution! Does not work in all cases.
*
* @param index the index
* @param type the type
* @param indexDefinition the index definition
* @param id the id
* @param source the source
* @return this
*/
BulkClient update(String index, String type, String id, String source);
BulkClient update(IndexDefinition indexDefinition, String id, String source);
/**
* Bulked update request. Each request will be added to a queue for bulking requests.
@ -111,42 +121,6 @@ public interface BulkClient extends BasicClient, Flushable {
*/
BulkClient update(UpdateRequest updateRequest);
/**
* Start bulk mode for indexes.
* @param indexDefinition index definition
* @throws IOException if bulk could not be started
*/
void startBulk(IndexDefinition indexDefinition) throws IOException;
/**
* Start bulk mode.
*
* @param index index
* @param startRefreshIntervalSeconds refresh interval before bulk
* @param stopRefreshIntervalSeconds refresh interval after bulk
* @throws IOException if bulk could not be started
*/
void startBulk(String index, long startRefreshIntervalSeconds,
long stopRefreshIntervalSeconds) throws IOException;
/**
* Stop bulk mode.
*
* @param indexDefinition index definition
* @throws IOException if bulk could not be startet
*/
void stopBulk(IndexDefinition indexDefinition) throws IOException;
/**
* Stops bulk mode.
*
* @param index index
* @param timeout maximum wait time
* @param timeUnit time unit for timeout
* @throws IOException if bulk could not be stopped
*/
void stopBulk(String index, long timeout, TimeUnit timeUnit) throws IOException;
/**
* Wait for all outstanding bulk responses.
*
@ -170,14 +144,14 @@ public interface BulkClient extends BasicClient, Flushable {
/**
* Refresh the index.
*
* @param index index
* @param indexDefinition index definition
*/
void refreshIndex(String index);
void refreshIndex(IndexDefinition indexDefinition);
/**
* Flush the index. The cluster clears cache and completes indexing.
*
* @param index index
* @param indexDefinition index definition
*/
void flushIndex(String index);
void flushIndex(IndexDefinition indexDefinition);
}

5
elx-api/src/main/java/org/xbib/elx/api/BulkController.java
Unescape Escape View File

@ -22,9 +22,6 @@ public interface BulkController extends Closeable, Flushable {
void startBulkMode(IndexDefinition indexDefinition) throws IOException;
void startBulkMode(String indexName, long startRefreshIntervalInSeconds,
long stopRefreshIntervalInSeconds) throws IOException;
void bulkIndex(IndexRequest indexRequest);
void bulkDelete(DeleteRequest deleteRequest);
@ -35,6 +32,4 @@ public interface BulkController extends Closeable, Flushable {
void stopBulkMode(IndexDefinition indexDefinition) throws IOException;
void stopBulkMode(String index, long timeout, TimeUnit timeUnit) throws IOException;
}

16
elx-api/src/main/java/org/xbib/elx/api/IndexDefinition.java
Unescape Escape View File

@ -34,6 +34,14 @@ public interface IndexDefinition {
Pattern getDateTimePattern();
IndexDefinition setStartBulkRefreshSeconds(int seconds);
int getStartBulkRefreshSeconds();
IndexDefinition setStopBulkRefreshSeconds(int seconds);
int getStopBulkRefreshSeconds();
IndexDefinition setEnabled(boolean enabled);
boolean isEnabled();
@ -67,12 +75,4 @@ public interface IndexDefinition {
long getMaxWaitTime();
TimeUnit getMaxWaitTimeUnit();
IndexDefinition setStartRefreshInterval(long seconds);
long getStartRefreshInterval();
IndexDefinition setStopRefreshInterval(long seconds);
long getStopRefreshInterval();
}

195
elx-common/src/main/java/org/xbib/elx/common/AbstractAdminClient.java
Unescape Escape View File

@ -41,11 +41,6 @@ import org.elasticsearch.cluster.metadata.IndexMetaData;
import org.elasticsearch.cluster.metadata.MappingMetaData;
import org.elasticsearch.common.collect.ImmutableOpenMap;
import org.elasticsearch.common.settings.Settings;
import org.elasticsearch.common.unit.TimeValue;
import org.elasticsearch.common.xcontent.ToXContent;
import org.elasticsearch.common.xcontent.XContentBuilder;
import org.elasticsearch.common.xcontent.json.JsonXContent;
import org.elasticsearch.common.xcontent.yaml.YamlXContent;
import org.elasticsearch.index.query.QueryBuilder;
import org.elasticsearch.index.query.QueryBuilders;
import org.elasticsearch.search.SearchHit;
@ -57,18 +52,10 @@ import org.xbib.elx.api.AdminClient;
import org.xbib.elx.api.IndexAliasAdder;
import org.xbib.elx.api.IndexDefinition;
import org.xbib.elx.api.IndexPruneResult;
import org.xbib.elx.api.IndexRetention;
import org.xbib.elx.api.IndexShiftResult;
import java.io.IOException;
import java.io.InputStream;
import java.net.MalformedURLException;
import java.net.URL;
import java.nio.charset.MalformedInputException;
import java.nio.charset.StandardCharsets;
import java.time.LocalDate;
import java.time.ZoneId;
import java.time.format.DateTimeFormatter;
import java.util.ArrayList;
import java.util.Collection;
import java.util.Collections;
@ -94,30 +81,28 @@ public abstract class AbstractAdminClient extends AbstractBasicClient implements
private static final Logger logger = LogManager.getLogger(AbstractAdminClient.class.getName());
/**
* The one and only index type name used in the extended client.
* Notr that all Elasticsearch version < 6.2.0 do not allow a prepending "_".
*/
private static final String TYPE_NAME = "doc";
@Override
public Map<String, ?> getMapping(String index) throws IOException {
public Map<String, ?> getMapping(IndexDefinition indexDefinition) throws IOException {
if (!ensureIndexDefinition(indexDefinition)) {
return null;
}
GetMappingsRequestBuilder getMappingsRequestBuilder = new GetMappingsRequestBuilder(client, GetMappingsAction.INSTANCE)
.setIndices(index)
.setTypes(TYPE_NAME);
.setIndices(indexDefinition.getFullIndexName())
.setTypes(indexDefinition.getType());
GetMappingsResponse getMappingsResponse = getMappingsRequestBuilder.execute().actionGet();
logger.info("get mappings response = {}", getMappingsResponse.getMappings().get(index).get(TYPE_NAME).getSourceAsMap());
return getMappingsResponse.getMappings().get(index).get(TYPE_NAME).getSourceAsMap();
return getMappingsResponse.getMappings()
.get(indexDefinition.getFullIndexName())
.get(indexDefinition.getType())
.getSourceAsMap();
}
@Override
public AdminClient deleteIndex(IndexDefinition indexDefinition) {
return deleteIndex(indexDefinition.getFullIndexName());
}
@Override
public AdminClient deleteIndex(String index) {
if (!ensureIndexDefinition(indexDefinition)) {
return null;
}
ensureClientIsPresent();
String index = indexDefinition.getFullIndexName();
if (index == null) {
logger.warn("no index name given to delete index");
return this;
@ -130,27 +115,27 @@ public abstract class AbstractAdminClient extends AbstractBasicClient implements
@Override
public AdminClient updateReplicaLevel(IndexDefinition indexDefinition, int level) throws IOException {
return updateReplicaLevel(indexDefinition.getFullIndexName(), level,
indexDefinition.getMaxWaitTime(), indexDefinition.getMaxWaitTimeUnit());
}
@Override
public AdminClient updateReplicaLevel(String index, int level, long maxWaitTime, TimeUnit timeUnit) throws IOException {
if (!ensureIndexDefinition(indexDefinition)) {
return null;
}
if (level < 1) {
logger.warn("invalid replica level");
return this;
}
String index = indexDefinition.getFullIndexName();
long maxWaitTime = indexDefinition.getMaxWaitTime();
TimeUnit timeUnit = indexDefinition.getMaxWaitTimeUnit();
updateIndexSetting(index, "number_of_replicas", level, maxWaitTime, timeUnit);
return this;
}
@Override
public int getReplicaLevel(IndexDefinition indexDefinition) {
return getReplicaLevel(indexDefinition.getFullIndexName());
}
@Override
public int getReplicaLevel(String index) {
if (!ensureIndexDefinition(indexDefinition)) {
return -1;
}
ensureClientIsPresent();
String index = indexDefinition.getFullIndexName();
GetSettingsRequest request = new GetSettingsRequest().indices(index);
GetSettingsResponse response = client.execute(GetSettingsAction.INSTANCE, request).actionGet();
int replica = -1;
@ -165,10 +150,10 @@ public abstract class AbstractAdminClient extends AbstractBasicClient implements
@Override
public String resolveMostRecentIndex(String alias) {
ensureClientIsPresent();
if (alias == null) {
return null;
}
ensureClientIsPresent();
GetAliasesRequest getAliasesRequest = new GetAliasesRequest().aliases(alias);
GetAliasesResponse getAliasesResponse = client.execute(GetAliasesAction.INSTANCE, getAliasesRequest).actionGet();
Pattern pattern = Pattern.compile("^(.*?)(\\d+)$");
@ -187,6 +172,7 @@ public abstract class AbstractAdminClient extends AbstractBasicClient implements
if (index == null) {
return Collections.emptyMap();
}
ensureClientIsPresent();
GetAliasesRequest getAliasesRequest = new GetAliasesRequest().indices(index);
return getFilters(client.execute(GetAliasesAction.INSTANCE, getAliasesRequest).actionGet());
}
@ -210,12 +196,6 @@ public abstract class AbstractAdminClient extends AbstractBasicClient implements
return aliasOrIndex != null ? aliasOrIndex.getIndices().stream().map(IndexMetaData::getIndex).collect(Collectors.toList()) : Collections.emptyList();
}
@Override
public IndexShiftResult shiftIndex(IndexDefinition indexDefinition,
List<String> additionalAliases) {
return shiftIndex(indexDefinition, additionalAliases, null);
}
@Override
public IndexShiftResult shiftIndex(IndexDefinition indexDefinition,
List<String> additionalAliases,
@ -223,6 +203,9 @@ public abstract class AbstractAdminClient extends AbstractBasicClient implements
if (additionalAliases == null) {
return new EmptyIndexShiftResult();
}
if (!ensureIndexDefinition(indexDefinition)) {
return new EmptyIndexShiftResult();
}
if (indexDefinition.isShiftEnabled()) {
return shiftIndex(indexDefinition.getIndex(),
indexDefinition.getFullIndexName(), additionalAliases.stream()
@ -380,7 +363,10 @@ public abstract class AbstractAdminClient extends AbstractBasicClient implements
}
@Override
public Long mostRecentDocument(String index, String timestampfieldname) {
public Long mostRecentDocument(IndexDefinition indexDefinition, String timestampfieldname) {
if (!ensureIndexDefinition(indexDefinition)) {
return -1L;
}
ensureClientIsPresent();
SortBuilder sort = SortBuilders.fieldSort(timestampfieldname).order(SortOrder.DESC);
SearchSourceBuilder builder = new SearchSourceBuilder();
@ -388,7 +374,7 @@ public abstract class AbstractAdminClient extends AbstractBasicClient implements
builder.field(timestampfieldname);
builder.size(1);
SearchRequest searchRequest = new SearchRequest();
searchRequest.indices(index);
searchRequest.indices(indexDefinition.getFullIndexName());
searchRequest.source(builder);
SearchResponse searchResponse = client.execute(SearchAction.INSTANCE, searchRequest).actionGet();
if (searchResponse.getHits().getHits().length == 1) {
@ -399,25 +385,25 @@ public abstract class AbstractAdminClient extends AbstractBasicClient implements
return 0L;
}
}
return null;
// almost impossible
return -1L;
}
@Override
public boolean forceMerge(IndexDefinition indexDefinition) {
if (indexDefinition.isForceMergeEnabled()) {
return forceMerge(indexDefinition.getFullIndexName(), indexDefinition.getMaxWaitTime(),
indexDefinition.getMaxWaitTimeUnit());
if (!ensureIndexDefinition(indexDefinition)) {
return false;
}
return false;
}
@Override
public boolean forceMerge(String index, long maxWaitTime, TimeUnit timeUnit) {
TimeValue timeout = toTimeValue(maxWaitTime, timeUnit);
if (!indexDefinition.isForceMergeEnabled()) {
return false;
}
ensureClientIsPresent();
String index = indexDefinition.getFullIndexName();
ForceMergeRequest forceMergeRequest = new ForceMergeRequest();
forceMergeRequest.indices(index);
try {
client.execute(ForceMergeAction.INSTANCE, forceMergeRequest).get(timeout.getMillis(), TimeUnit.MILLISECONDS);
client.execute(ForceMergeAction.INSTANCE, forceMergeRequest)
.get(indexDefinition.getMaxWaitTime(), indexDefinition.getMaxWaitTimeUnit());
return true;
} catch (TimeoutException e) {
logger.error("timeout");
@ -430,45 +416,12 @@ public abstract class AbstractAdminClient extends AbstractBasicClient implements
return false;
}
@Override
public IndexDefinition buildIndexDefinitionFromSettings(String index, Settings settings)
throws IOException {
boolean isEnabled = settings.getAsBoolean("enabled", false);
String indexName = settings.get("name", index);
String dateTimePatternStr = settings.get("dateTimePattern", "^(.*?)(\\\\d+)$");
Pattern dateTimePattern = Pattern.compile(dateTimePatternStr);
String dateTimeFormat = settings.get("dateTimeFormat", "yyyyMMdd");
DateTimeFormatter dateTimeFormatter = DateTimeFormatter.ofPattern(dateTimeFormat)
.withZone(ZoneId.systemDefault());
String fullName = indexName + dateTimeFormatter.format(LocalDate.now());
String fullIndexName = resolveAlias(fullName).stream().findFirst().orElse(fullName);
IndexRetention indexRetention = new DefaultIndexRetention()
.setMinToKeep(settings.getAsInt("retention.mintokeep", 0))
.setDelta(settings.getAsInt("retention.delta", 0));
return new DefaultIndexDefinition()
.setEnabled(isEnabled)
.setIndex(indexName)
.setFullIndexName(fullIndexName)
.setSettings(findSettingsFrom(settings.get("settings")))
.setMappings(findMappingsFrom(settings.get("mapping")))
.setDateTimeFormatter(dateTimeFormatter)
.setDateTimePattern(dateTimePattern)
.setIgnoreErrors(settings.getAsBoolean("skiperrors", false))
.setShift(settings.getAsBoolean("shift", true))
.setPrune(settings.getAsBoolean("prune", true))
.setReplicaLevel(settings.getAsInt("replica", 0))
.setMaxWaitTime(settings.getAsLong("timeout", 30L), TimeUnit.SECONDS)
.setRetention(indexRetention)
.setStartRefreshInterval(settings.getAsLong("bulk.startrefreshinterval", -1L))
.setStopRefreshInterval(settings.getAsLong("bulk.stoprefreshinterval", -1L));
}
@Override
public void updateIndexSetting(String index, String key, Object value, long timeout, TimeUnit timeUnit) throws IOException {
ensureClientIsPresent();
if (index == null) {
throw new IOException("no index name given");
}
ensureClientIsPresent();
Settings.Builder updateSettingsBuilder = Settings.builder();
updateSettingsBuilder.put(key, value.toString());
UpdateSettingsRequest updateSettingsRequest = new UpdateSettingsRequest(index)
@ -477,9 +430,12 @@ public abstract class AbstractAdminClient extends AbstractBasicClient implements
}
@Override
public void checkMapping(String index) {
public void checkMapping(IndexDefinition indexDefinition) {
if (!ensureIndexDefinition(indexDefinition)) {
return;
}
ensureClientIsPresent();
GetMappingsRequest getMappingsRequest = new GetMappingsRequest().indices(index);
GetMappingsRequest getMappingsRequest = new GetMappingsRequest().indices(indexDefinition.getFullIndexName());
GetMappingsResponse getMappingsResponse = client.execute(GetMappingsAction.INSTANCE, getMappingsRequest).actionGet();
ImmutableOpenMap<String, ImmutableOpenMap<String, MappingMetaData>> map = getMappingsResponse.getMappings();
map.keys().forEach((Consumer<ObjectCursor<String>>) stringObjectCursor -> {
@ -487,54 +443,11 @@ public abstract class AbstractAdminClient extends AbstractBasicClient implements
for (ObjectObjectCursor<String, MappingMetaData> cursor : mappings) {
String mappingName = cursor.key;
MappingMetaData mappingMetaData = cursor.value;
checkMapping(index, mappingName, mappingMetaData);
checkMapping(indexDefinition.getFullIndexName(), mappingName, mappingMetaData);
}
});
}
private static String findSettingsFrom(String string) throws IOException {
if (string == null) {
return null;
}
try {
URL url = new URL(string);
try (InputStream inputStream = url.openStream()) {
Settings settings = Settings.builder().loadFromStream(string, inputStream).build();
XContentBuilder builder = JsonXContent.contentBuilder();
settings.toXContent(builder, ToXContent.EMPTY_PARAMS);
return builder.string();
}
} catch (MalformedURLException e) {
return string;
}
}
private static String findMappingsFrom(String string) throws IOException {
if (string == null) {
return null;
}
try {
URL url = new URL(string);
try (InputStream inputStream = url.openStream()) {
if (string.endsWith(".json")) {
Map<String, ?> mappings = JsonXContent.jsonXContent.createParser(inputStream).mapOrdered();
XContentBuilder builder = JsonXContent.contentBuilder();
builder.startObject().map(mappings).endObject();
return builder.string();
}
if (string.endsWith(".yml") || string.endsWith(".yaml")) {
Map<String, ?> mappings = YamlXContent.yamlXContent.createParser(inputStream).mapOrdered();
XContentBuilder builder = JsonXContent.contentBuilder();
builder.startObject().map(mappings).endObject();
return builder.string();
}
}
return string;
} catch (MalformedInputException e) {
return string;
}
}
private Map<String, String> getFilters(GetAliasesResponse getAliasesResponse) {
Map<String, String> result = new HashMap<>();
for (ObjectObjectCursor<String, List<AliasMetaData>> object : getAliasesResponse.getAliases()) {
@ -654,7 +567,6 @@ public abstract class AbstractAdminClient extends AbstractBasicClient implements
}
}
private static class EmptyIndexShiftResult implements IndexShiftResult {
@Override
@ -774,5 +686,4 @@ public abstract class AbstractAdminClient extends AbstractBasicClient implements
return "EMPTY PRUNE";
}
}
}

28
elx-common/src/main/java/org/xbib/elx/common/AbstractBasicClient.java
Unescape Escape View File

@ -24,6 +24,7 @@ import org.elasticsearch.common.settings.Settings;
import org.elasticsearch.common.unit.TimeValue;
import org.elasticsearch.index.query.QueryBuilders;
import org.xbib.elx.api.BasicClient;
import org.xbib.elx.api.IndexDefinition;
import java.io.IOException;
import java.util.concurrent.TimeUnit;
import java.util.concurrent.atomic.AtomicBoolean;
@ -32,12 +33,6 @@ public abstract class AbstractBasicClient implements BasicClient {
private static final Logger logger = LogManager.getLogger(AbstractBasicClient.class.getName());
/**
* The one and only index type name used in the extended client.
* Note that all Elasticsearch versions before 6.2.0 do not allow a prepending "_".
*/
protected static final String TYPE_NAME = "doc";
protected ElasticsearchClient client;
protected Settings settings;
@ -144,18 +139,23 @@ public abstract class AbstractBasicClient implements BasicClient {
}
@Override
public long getSearchableDocs(String index) {
public long getSearchableDocs(IndexDefinition indexDefinition) {
if (!ensureIndexDefinition(indexDefinition)) {
return -1L;
}
ensureClientIsPresent();
SearchRequestBuilder searchRequestBuilder = new SearchRequestBuilder(client, SearchAction.INSTANCE)
.setIndices(index)
.setIndices(indexDefinition.getFullIndexName())
.setQuery(QueryBuilders.matchAllQuery())
.setSize(0);
return searchRequestBuilder.execute().actionGet().getHits().getTotalHits();
}
@Override
public boolean isIndexExists(String index) {
public boolean isIndexExists(IndexDefinition indexDefinition) {
ensureClientIsPresent();
IndicesExistsRequest indicesExistsRequest = new IndicesExistsRequest();
indicesExistsRequest.indices(new String[]{index});
indicesExistsRequest.indices(new String[] { indexDefinition.getFullIndexName() } );
IndicesExistsResponse indicesExistsResponse =
client.execute(IndicesExistsAction.INSTANCE, indicesExistsRequest).actionGet();
return indicesExistsResponse.isExists();
@ -198,6 +198,14 @@ public abstract class AbstractBasicClient implements BasicClient {
}
}
protected boolean ensureIndexDefinition(IndexDefinition indexDefinition) {
if (!indexDefinition.isEnabled()) {
logger.warn("index " + indexDefinition.getFullIndexName() + " is disabled");
return false;
}
return true;
}
protected static TimeValue toTimeValue(long timeValue, TimeUnit timeUnit) {
switch (timeUnit) {
case DAYS:

111
elx-common/src/main/java/org/xbib/elx/common/AbstractBulkClient.java
Unescape Escape View File

@ -1,5 +1,6 @@
package org.xbib.elx.common;
import com.google.common.base.Charsets;
import org.apache.logging.log4j.Level;
import org.apache.logging.log4j.LogManager;
import org.apache.logging.log4j.Logger;
@ -16,6 +17,7 @@ import org.elasticsearch.action.update.UpdateRequest;
import org.elasticsearch.common.bytes.BytesArray;
import org.elasticsearch.common.bytes.BytesReference;
import org.elasticsearch.common.settings.Settings;
import org.elasticsearch.common.xcontent.XContentBuilder;
import org.elasticsearch.common.xcontent.json.JsonXContent;
import org.xbib.elx.api.BulkClient;
import org.xbib.elx.api.BulkController;
@ -70,6 +72,9 @@ public abstract class AbstractBulkClient extends AbstractBasicClient implements
@Override
public void newIndex(IndexDefinition indexDefinition) throws IOException {
if (!ensureIndexDefinition(indexDefinition)) {
return;
}
String index = indexDefinition.getFullIndexName();
if (index == null) {
throw new IllegalArgumentException("no index name given");
@ -81,16 +86,22 @@ public abstract class AbstractBulkClient extends AbstractBasicClient implements
ensureClientIsPresent();
CreateIndexRequestBuilder createIndexRequestBuilder = new CreateIndexRequestBuilder(client, CreateIndexAction.INSTANCE)
.setIndex(index);
Settings settings = indexDefinition.getSettings() == null ? null :
Settings.builder().loadFromSource(indexDefinition.getSettings()).build();
if (settings != null) {
createIndexRequestBuilder.setSettings(settings);
if (indexDefinition.getSettings() == null) {
XContentBuilder builder = JsonXContent.contentBuilder()
.startObject()
.startObject("index")
.field("number_of_shards", 1)
.field("number_of_replicas", 0)
.endObject()
.endObject();
indexDefinition.setSettings(builder.string());
}
Settings settings = Settings.builder().loadFromSource(indexDefinition.getSettings()).build();
createIndexRequestBuilder.setSettings(settings);
// must be Map<String, Object> to match prototype of addMapping()!
Map<String, Object> mappings = indexDefinition.getMappings() == null ? null :
JsonXContent.jsonXContent.createParser(indexDefinition.getMappings()).mapOrdered();
if (mappings != null) {
logger.info("mappings = " + mappings);
createIndexRequestBuilder.addMapping(type, mappings);
} else {
createIndexRequestBuilder.addMapping(type,
@ -101,52 +112,45 @@ public abstract class AbstractBulkClient extends AbstractBasicClient implements
logger.info("index {} created", index);
} else {
logger.warn("index creation of {} not acknowledged", index);
return;
}
refreshIndex(index);
// we really need state GREEN. If yellow, we may trigger shard write errors and queue will exceed quickly.
logger.info("waiting for GREEN after index {} was created", index);
waitForCluster("GREEN", indexDefinition.getMaxWaitTime(), indexDefinition.getMaxWaitTimeUnit());
}
@Override
public void startBulk(IndexDefinition indexDefinition) throws IOException {
startBulk(indexDefinition.getFullIndexName(), -1, 1);
}
@Override
public void startBulk(String index, long startRefreshIntervalSeconds, long stopRefreshIntervalSeconds)
throws IOException {
if (!ensureIndexDefinition(indexDefinition)) {
return;
}
ensureClientIsPresent();
bulkController.startBulkMode(index, startRefreshIntervalSeconds, stopRefreshIntervalSeconds);
bulkController.startBulkMode(indexDefinition);
}
@Override
public void stopBulk(IndexDefinition indexDefinition) throws IOException {
if (!ensureIndexDefinition(indexDefinition)) {
return;
}
ensureClientIsPresent();
bulkController.stopBulkMode(indexDefinition);
}
@Override
public void stopBulk(String index, long timeout, TimeUnit timeUnit) throws IOException {
ensureClientIsPresent();
bulkController.stopBulkMode(index, timeout, timeUnit);
}
@Override
public BulkClient index(String index, String type, String id, boolean create, String source) {
return index(new IndexRequest()
.index(index)
.type(type)
.id(id)
.create(create)
.source(source)); // will be converted into a bytes reference
public BulkClient index(IndexDefinition indexDefinition, String id, boolean create, String source) {
return index(indexDefinition, id, create, new BytesArray(source.getBytes(Charsets.UTF_8)));
}
@Override
public BulkClient index(String index, String type, String id, boolean create, BytesReference source) {
public BulkClient index(IndexDefinition indexDefinition, String id, boolean create, BytesReference source) {
if (!ensureIndexDefinition(indexDefinition)) {
return this;
}
return index(new IndexRequest()
.index(index)
.type(type)
.id(id)
.create(create)
.source(source));
.index(indexDefinition.getFullIndexName())
.type(indexDefinition.getType())
.id(id).create(create).source(source));
}
@Override
@ -157,10 +161,13 @@ public abstract class AbstractBulkClient extends AbstractBasicClient implements
}
@Override
public BulkClient delete(String index, String type, String id) {
public BulkClient delete(IndexDefinition indexDefinition, String id) {
if (!ensureIndexDefinition(indexDefinition)) {
return this;
}
return delete(new DeleteRequest()
.index(index)
.type(type)
.index(indexDefinition.getFullIndexName())
.type(indexDefinition.getType())
.id(id));
}
@ -172,17 +179,19 @@ public abstract class AbstractBulkClient extends AbstractBasicClient implements
}
@Override
public BulkClient update(String index, String type, String id, String source) {
return update(index, type, id, new BytesArray(source.getBytes(StandardCharsets.UTF_8)));
public BulkClient update(IndexDefinition indexDefinition, String id, String source) {
return update(indexDefinition, id, new BytesArray(source.getBytes(StandardCharsets.UTF_8)));
}
@Override
public BulkClient update(String index, String type, String id, BytesReference source) {
public BulkClient update(IndexDefinition indexDefinition, String id, BytesReference source) {
if (!ensureIndexDefinition(indexDefinition)) {
return this;
}
return update(new UpdateRequest()
.index(index)
.type(type)
.id(id)
.doc(source.hasArray() ? source.array() : source.toBytes()));
.index(indexDefinition.getFullIndexName())
.type(indexDefinition.getType())
.id(id).doc(source.hasArray() ? source.array() : source.toBytes()));
}
@Override
@ -204,18 +213,20 @@ public abstract class AbstractBulkClient extends AbstractBasicClient implements
}
@Override
public void flushIndex(String index) {
if (index != null) {
ensureClientIsPresent();
client.execute(FlushAction.INSTANCE, new FlushRequest(index)).actionGet();
public void flushIndex(IndexDefinition indexDefinition) {
if (!ensureIndexDefinition(indexDefinition)) {
return;
}
ensureClientIsPresent();
client.execute(FlushAction.INSTANCE, new FlushRequest(indexDefinition.getFullIndexName())).actionGet();
}
@Override
public void refreshIndex(String index) {
if (index != null) {
ensureClientIsPresent();
client.execute(RefreshAction.INSTANCE, new RefreshRequest(index)).actionGet();
public void refreshIndex(IndexDefinition indexDefinition) {
if (!ensureIndexDefinition(indexDefinition)) {
return;
}
ensureClientIsPresent();
client.execute(RefreshAction.INSTANCE, new RefreshRequest(indexDefinition.getFullIndexName())).actionGet();
}
}

16
elx-common/src/main/java/org/xbib/elx/common/ClientBuilder.java
Unescape Escape View File

@ -47,6 +47,17 @@ public class ClientBuilder {
this.classLoader = classLoader;
this.settingsBuilder = Settings.builder();
settingsBuilder.put("node.name", "elx-client-" + Version.CURRENT);
for (Parameters p : Parameters.values()) {
if (p.getType() == Boolean.class) {
settingsBuilder.put(p.getName(), p.getBoolean());
}
if (p.getType() == Integer.class) {
settingsBuilder.put(p.getName(), p.getInteger());
}
if (p.getType() == String.class) {
settingsBuilder.put(p.getName(), p.getString());
}
}
}
public static ClientBuilder builder() {
@ -82,6 +93,11 @@ public class ClientBuilder {
return this;
}
public ClientBuilder put(String key, Boolean value) {
settingsBuilder.put(key, value);
return this;
}
public ClientBuilder put(String key, Long value) {
settingsBuilder.put(key, value);
return this;

112
elx-common/src/main/java/org/xbib/elx/common/DefaultBulkController.java
Unescape Escape View File

@ -16,10 +16,6 @@ import org.xbib.elx.api.BulkProcessor;
import org.xbib.elx.api.IndexDefinition;
import java.io.IOException;
import java.util.ArrayList;
import java.util.HashMap;
import java.util.List;
import java.util.Map;
import java.util.concurrent.TimeUnit;
import java.util.concurrent.atomic.AtomicBoolean;
@ -33,11 +29,7 @@ public class DefaultBulkController implements BulkController {
private BulkProcessor bulkProcessor;
private final List<String> indexNames;
private final Map<String, Long> startBulkRefreshIntervals;
private final Map<String, Long> stopBulkRefreshIntervals;
private BulkListener bulkListener;
private final long maxWaitTime;
@ -48,10 +40,7 @@ public class DefaultBulkController implements BulkController {
public DefaultBulkController(BulkClient bulkClient) {
this.bulkClient = bulkClient;
this.bulkMetric = new DefaultBulkMetric();
this.indexNames = new ArrayList<>();
this.active = new AtomicBoolean(false);
this.startBulkRefreshIntervals = new HashMap<>();
this.stopBulkRefreshIntervals = new HashMap<>();
this.maxWaitTime = 30L;
this.maxWaitTimeUnit = TimeUnit.SECONDS;
}
@ -63,24 +52,27 @@ public class DefaultBulkController implements BulkController {
@Override
public Throwable getLastBulkError() {
return bulkProcessor.getBulkListener().getLastBulkError();
return bulkListener.getLastBulkError();
}
@Override
public void init(Settings settings) {
bulkMetric.init(settings);
int maxActionsPerRequest = settings.getAsInt(Parameters.MAX_ACTIONS_PER_REQUEST.name(),
Parameters.DEFAULT_MAX_ACTIONS_PER_REQUEST.getNum());
int maxConcurrentRequests = settings.getAsInt(Parameters.MAX_CONCURRENT_REQUESTS.name(),
Parameters.DEFAULT_MAX_CONCURRENT_REQUESTS.getNum());
TimeValue flushIngestInterval = settings.getAsTime(Parameters.FLUSH_INTERVAL.name(),
TimeValue.timeValueSeconds(Parameters.DEFAULT_FLUSH_INTERVAL.getNum()));
ByteSizeValue maxVolumePerRequest = settings.getAsBytesSize(Parameters.MAX_VOLUME_PER_REQUEST.name(),
ByteSizeValue.parseBytesSizeValue(Parameters.DEFAULT_MAX_VOLUME_PER_REQUEST.getString(),
"maxVolumePerRequest"));
boolean enableBulkLogging = settings.getAsBoolean(Parameters.ENABLE_BULK_LOGGING.name(),
Parameters.ENABLE_BULK_LOGGING.getValue());
BulkListener bulkListener = new DefaultBulkListener(this, bulkMetric, enableBulkLogging);
int maxActionsPerRequest = settings.getAsInt(Parameters.MAX_ACTIONS_PER_REQUEST.getName(),
Parameters.MAX_ACTIONS_PER_REQUEST.getInteger());
int maxConcurrentRequests = settings.getAsInt(Parameters.MAX_CONCURRENT_REQUESTS.getName(),
Parameters.MAX_CONCURRENT_REQUESTS.getInteger());
String flushIngestIntervalStr = settings.get(Parameters.FLUSH_INTERVAL.getName(),
Parameters.FLUSH_INTERVAL.getString());
TimeValue flushIngestInterval = TimeValue.parseTimeValue(flushIngestIntervalStr,
TimeValue.timeValueSeconds(30), "");
ByteSizeValue maxVolumePerRequest = settings.getAsBytesSize(Parameters.MAX_VOLUME_PER_REQUEST.getName(),
ByteSizeValue.parseBytesSizeValue(Parameters.MAX_VOLUME_PER_REQUEST.getString(), "1m"));
boolean enableBulkLogging = settings.getAsBoolean(Parameters.ENABLE_BULK_LOGGING.getName(),
Parameters.ENABLE_BULK_LOGGING.getBoolean());
boolean failOnBulkError = settings.getAsBoolean(Parameters.FAIL_ON_BULK_ERROR.getName(),
Parameters.FAIL_ON_BULK_ERROR.getBoolean());
this.bulkListener = new DefaultBulkListener(this, bulkMetric, enableBulkLogging, failOnBulkError);
this.bulkProcessor = DefaultBulkProcessor.builder(bulkClient.getClient(), bulkListener)
.setBulkActions(maxActionsPerRequest)
.setConcurrentRequests(maxConcurrentRequests)
@ -90,9 +82,13 @@ public class DefaultBulkController implements BulkController {
this.active.set(true);
if (logger.isInfoEnabled()) {
logger.info("bulk processor up with maxActionsPerRequest = {} maxConcurrentRequests = {} " +
"flushIngestInterval = {} maxVolumePerRequest = {} bulk logging = {} logger debug = {} from settings = {}",
maxActionsPerRequest, maxConcurrentRequests, flushIngestInterval, maxVolumePerRequest,
enableBulkLogging, logger.isDebugEnabled(), settings.toDelimitedString(','));
"flushIngestInterval = {} maxVolumePerRequest = {} " +
"bulk logging = {} fail on bulk error = {} " +
"logger debug = {} from settings = {}",
maxActionsPerRequest, maxConcurrentRequests,
flushIngestInterval, maxVolumePerRequest,
enableBulkLogging, failOnBulkError,
logger.isDebugEnabled(), settings.toDelimitedString(','));
}
}
@ -103,22 +99,11 @@ public class DefaultBulkController implements BulkController {
@Override
public void startBulkMode(IndexDefinition indexDefinition) throws IOException {
startBulkMode(indexDefinition.getFullIndexName(), indexDefinition.getStartRefreshInterval(),
indexDefinition.getStopRefreshInterval());
}
@Override
public void startBulkMode(String indexName,
long startRefreshIntervalInSeconds,
long stopRefreshIntervalInSeconds) throws IOException {
if (!indexNames.contains(indexName)) {
indexNames.add(indexName);
startBulkRefreshIntervals.put(indexName, startRefreshIntervalInSeconds);
stopBulkRefreshIntervals.put(indexName, stopRefreshIntervalInSeconds);
if (startRefreshIntervalInSeconds != 0L) {
bulkClient.updateIndexSetting(indexName, "refresh_interval", startRefreshIntervalInSeconds + "s",
30L, TimeUnit.SECONDS);
}
String indexName = indexDefinition.getFullIndexName();
if (indexDefinition.getStartBulkRefreshSeconds() != 0) {
bulkClient.updateIndexSetting(indexName, "refresh_interval",
indexDefinition.getStartBulkRefreshSeconds() + "s",
30L, TimeUnit.SECONDS);
}
}
@ -167,31 +152,30 @@ public class DefaultBulkController implements BulkController {
@Override
public boolean waitForBulkResponses(long timeout, TimeUnit timeUnit) {
try {
return bulkProcessor.awaitFlush(timeout, timeUnit);
if (bulkProcessor != null) {
bulkProcessor.flush();
return bulkProcessor.awaitFlush(timeout, timeUnit);
}
} catch (InterruptedException e) {
Thread.currentThread().interrupt();
logger.error("interrupted");
return false;
} catch (IOException e) {
logger.error(e.getMessage(), e);
return false;
}
return false;
}
@Override
public void stopBulkMode(IndexDefinition indexDefinition) throws IOException {
stopBulkMode(indexDefinition.getFullIndexName(),
indexDefinition.getMaxWaitTime(), indexDefinition.getMaxWaitTimeUnit());
}
@Override
public void stopBulkMode(String index, long timeout, TimeUnit timeUnit) throws IOException {
flush();
if (waitForBulkResponses(timeout, timeUnit)) {
if (indexNames.contains(index)) {
Long secs = stopBulkRefreshIntervals.get(index);
if (secs != null && secs != 0L) {
bulkClient.updateIndexSetting(index, "refresh_interval", secs + "s",
30L, TimeUnit.SECONDS);
}
indexNames.remove(index);
if (waitForBulkResponses(indexDefinition.getMaxWaitTime(), indexDefinition.getMaxWaitTimeUnit())) {
if (indexDefinition.getStopBulkRefreshSeconds() != 0) {
bulkClient.updateIndexSetting(indexDefinition.getFullIndexName(),
"refresh_interval",
indexDefinition.getStopBulkRefreshSeconds() + "s",
30L, TimeUnit.SECONDS);
}
}
}
@ -206,15 +190,7 @@ public class DefaultBulkController implements BulkController {
@Override
public void close() throws IOException {
flush();
if (bulkClient.waitForResponses(maxWaitTime, maxWaitTimeUnit)) {
for (String index : indexNames) {
Long secs = stopBulkRefreshIntervals.get(index);
if (secs != null && secs != 0L)
bulkClient.updateIndexSetting(index, "refresh_interval", secs + "s",
30L, TimeUnit.SECONDS);
}
indexNames.clear();
}
bulkClient.waitForResponses(maxWaitTime, maxWaitTimeUnit);
if (bulkProcessor != null) {
bulkProcessor.close();
}

10
elx-common/src/main/java/org/xbib/elx/common/DefaultBulkListener.java
Unescape Escape View File

@ -19,14 +19,18 @@ public class DefaultBulkListener implements BulkListener {
private final boolean isBulkLoggingEnabled;
private final boolean failOnError;
private Throwable lastBulkError = null;
public DefaultBulkListener(BulkController bulkController,
BulkMetric bulkMetric,
boolean isBulkLoggingEnabled) {
boolean isBulkLoggingEnabled,
boolean failOnError) {
this.bulkController = bulkController;
this.bulkMetric = bulkMetric;
this.isBulkLoggingEnabled = isBulkLoggingEnabled;
this.failOnError = failOnError;
}
@Override
@ -73,6 +77,10 @@ public class DefaultBulkListener implements BulkListener {
logger.error("bulk [{}] failed with {} failed items, failure message = {}",
executionId, n, response.buildFailureMessage());
}
if (failOnError) {
throw new IllegalStateException("bulk failed: id = " + executionId +
" n = " + n + " message = " + response.buildFailureMessage());
}
} else {
bulkMetric.getCurrentIngestNumDocs().dec(response.getItems().length);
}

140
elx-common/src/main/java/org/xbib/elx/common/DefaultIndexDefinition.java
Unescape Escape View File

@ -1,10 +1,24 @@
package org.xbib.elx.common;
import org.elasticsearch.common.settings.Settings;
import org.elasticsearch.common.xcontent.ToXContent;
import org.elasticsearch.common.xcontent.XContentBuilder;
import org.elasticsearch.common.xcontent.json.JsonXContent;
import org.elasticsearch.common.xcontent.yaml.YamlXContent;
import org.xbib.elx.api.AdminClient;
import org.xbib.elx.api.IndexDefinition;
import org.xbib.elx.api.IndexRetention;
import java.io.IOException;
import java.io.InputStream;
import java.net.MalformedURLException;
import java.net.URL;
import java.nio.charset.MalformedInputException;
import java.time.LocalDate;
import java.time.ZoneId;
import java.time.format.DateTimeFormatter;
import java.util.Locale;
import java.util.Map;
import java.util.concurrent.TimeUnit;
import java.util.regex.Pattern;
@ -42,13 +56,50 @@ public class DefaultIndexDefinition implements IndexDefinition {
private TimeUnit maxWaitTimeUnit;
private long startRefreshInterval;
private int startRefreshInterval;
private long stopRefreshInterval;
private int stopRefreshInterval;
public DefaultIndexDefinition() {
public DefaultIndexDefinition(String index, String type) {
setIndex(index);
setType(type);
setDateTimeFormatter(DateTimeFormatter.ofPattern("yyyyMMdd", Locale.getDefault()));
setDateTimePattern(Pattern.compile("^(.*?)(\\d+)$"));
setFullIndexName(index + getDateTimeFormatter().format(LocalDate.now()));
setEnabled(true);
setMaxWaitTime(Parameters.MAX_WAIT_BULK_RESPONSE_SECONDS.getInteger(), TimeUnit.SECONDS);
}
public DefaultIndexDefinition(AdminClient adminClient, String index, String type, Settings settings)
throws IOException {
boolean isEnabled = settings.getAsBoolean("enabled", true);
String indexName = settings.get("name", index);
String dateTimePatternStr = settings.get("dateTimePattern", "^(.*?)(\\\\d+)$");
Pattern dateTimePattern = Pattern.compile(dateTimePatternStr);
String dateTimeFormat = settings.get("dateTimeFormat", "yyyyMMdd");
DateTimeFormatter dateTimeFormatter = DateTimeFormatter.ofPattern(dateTimeFormat)
.withZone(ZoneId.systemDefault());
String fullName = indexName + dateTimeFormatter.format(LocalDate.now());
String fullIndexName = adminClient.resolveAlias(fullName).stream().findFirst().orElse(fullName);
IndexRetention indexRetention = new DefaultIndexRetention()
.setMinToKeep(settings.getAsInt("retention.mintokeep", 0))
.setDelta(settings.getAsInt("retention.delta", 0));
setEnabled(isEnabled)
.setIndex(indexName)
.setType(type)
.setFullIndexName(fullIndexName)
.setSettings(findSettingsFrom(settings.get("settings")))
.setMappings(findMappingsFrom(settings.get("mapping")))
.setDateTimeFormatter(dateTimeFormatter)
.setDateTimePattern(dateTimePattern)
.setIgnoreErrors(settings.getAsBoolean("skiperrors", false))
.setShift(settings.getAsBoolean("shift", true))
.setPrune(settings.getAsBoolean("prune", true))
.setReplicaLevel(settings.getAsInt("replica", 0))
.setMaxWaitTime(settings.getAsLong("timeout", 30L), TimeUnit.SECONDS)
.setRetention(indexRetention)
.setStartBulkRefreshSeconds(settings.getAsInt(Parameters.START_BULK_REFRESH_SECONDS.getName(), -1))
.setStopBulkRefreshSeconds(settings.getAsInt(Parameters.STOP_BULK_REFRESH_SECONDS.getName(), -1));
}
@Override
@ -129,6 +180,28 @@ public class DefaultIndexDefinition implements IndexDefinition {
return pattern;
}
@Override
public IndexDefinition setStartBulkRefreshSeconds(int seconds) {
this.startRefreshInterval = seconds;
return this;
}
@Override
public int getStartBulkRefreshSeconds() {
return startRefreshInterval;
}
@Override
public IndexDefinition setStopBulkRefreshSeconds(int seconds) {
this.stopRefreshInterval = seconds;
return this;
}
@Override
public int getStopBulkRefreshSeconds() {
return stopRefreshInterval;
}
@Override
public IndexDefinition setEnabled(boolean enabled) {
this.enabled = enabled;
@ -223,26 +296,47 @@ public class DefaultIndexDefinition implements IndexDefinition {
return maxWaitTimeUnit;
}
@Override
public IndexDefinition setStartRefreshInterval(long seconds) {
this.startRefreshInterval = seconds;
return this;
}
@Override
public long getStartRefreshInterval() {
return startRefreshInterval;
private static String findSettingsFrom(String string) throws IOException {
if (string == null) {
return null;
}
try {
URL url = new URL(string);
try (InputStream inputStream = url.openStream()) {
Settings settings = Settings.builder().loadFromStream(string, inputStream).build();
XContentBuilder builder = JsonXContent.contentBuilder();
settings.toXContent(builder, ToXContent.EMPTY_PARAMS);
return builder.string();
}
} catch (MalformedURLException e) {
return string;
}
}
private static String findMappingsFrom(String string) throws IOException {
if (string == null) {
return null;
}
try {
URL url = new URL(string);
try (InputStream inputStream = url.openStream()) {
if (string.endsWith(".json")) {
Map<String, ?> mappings = JsonXContent.jsonXContent.createParser(inputStream).mapOrdered();
XContentBuilder builder = JsonXContent.contentBuilder();
builder.startObject().map(mappings).endObject();
return builder.string();
}
if (string.endsWith(".yml") || string.endsWith(".yaml")) {
Map<String, ?> mappings = YamlXContent.yamlXContent.createParser(inputStream).mapOrdered();
XContentBuilder builder = JsonXContent.contentBuilder();
builder.startObject().map(mappings).endObject();
return builder.string();
}
}
return string;
} catch (MalformedInputException e) {
return string;
}
}
@Override
public IndexDefinition setStopRefreshInterval(long seconds) {
this.stopRefreshInterval = seconds;
return this;
}
@Override
public long getStopRefreshInterval() {
return stopRefreshInterval;
}
}

16
elx-common/src/main/java/org/xbib/elx/common/MockAdminClient.java
Unescape Escape View File

@ -28,28 +28,12 @@ public class MockAdminClient extends AbstractAdminClient {
protected void closeClient(Settings settings) {
}
@Override
public MockAdminClient deleteIndex(String index) {
return this;
}
@Override
public boolean forceMerge(String index, long maxWaitTime, TimeUnit timeUnit) {
return true;
}
@Override
public void waitForCluster(String healthColor, long timeValue, TimeUnit timeUnit) {
}
@Override
public void waitForShards(long maxWaitTime, TimeUnit timeUnit) {
}
@Override
public MockAdminClient updateReplicaLevel(String index, int level, long maxWaitTime, TimeUnit timeUnit) {
return this;
}
@Override

32
elx-common/src/main/java/org/xbib/elx/common/MockBulkClient.java
Unescape Escape View File

@ -39,21 +39,6 @@ public class MockBulkClient extends AbstractBulkClient {
protected void closeClient(Settings settings) {
}
@Override
public MockBulkClient index(String index, String type, String id, boolean create, String source) {
return this;
}
@Override
public MockBulkClient delete(String index, String type, String id) {
return this;
}
@Override
public MockBulkClient update(String index, String type, String id, String source) {
return this;
}
@Override
public MockBulkClient index(IndexRequest indexRequest) {
return this;
@ -69,28 +54,11 @@ public class MockBulkClient extends AbstractBulkClient {
return this;
}
@Override
public void startBulk(String index, long startRefreshInterval, long stopRefreshIterval) {
}
@Override
public void stopBulk(String index, long maxWaitTime, TimeUnit timeUnit) {
}
@Override
public boolean waitForResponses(long maxWaitTime, TimeUnit timeUnit) {
return true;
}
@Override
public void refreshIndex(String index) {
}
@Override
public void flushIndex(String index) {
}
@Override
public void flush() {
// nothing to do

51
elx-common/src/main/java/org/xbib/elx/common/Parameters.java
Unescape Escape View File

@ -2,51 +2,54 @@ package org.xbib.elx.common;
public enum Parameters {
ENABLE_BULK_LOGGING(false),
MAX_WAIT_BULK_RESPONSE_SECONDS("bulk.max_wait_response_seconds", Integer.class, 30),
DEFAULT_MAX_ACTIONS_PER_REQUEST(1000),
START_BULK_REFRESH_SECONDS("bulk.start_refresh_seconds", Integer.class, 0),
DEFAULT_MAX_CONCURRENT_REQUESTS(Runtime.getRuntime().availableProcessors()),
STOP_BULK_REFRESH_SECONDS("bulk.stop_refresh_seconds", Integer.class, 30),
DEFAULT_MAX_VOLUME_PER_REQUEST("10mb"),
ENABLE_BULK_LOGGING("bulk.logging.enabled", Boolean.class, true),
DEFAULT_FLUSH_INTERVAL(30),
FAIL_ON_BULK_ERROR("bulk.failonerror", Boolean.class, true),
MAX_ACTIONS_PER_REQUEST ("max_actions_per_request"),
MAX_ACTIONS_PER_REQUEST("bulk.max_actions_per_request", Integer.class, 1000),
MAX_CONCURRENT_REQUESTS("max_concurrent_requests"),
// 0 = 1 CPU, synchronous requests, &gt; 0 = n + 1 CPUs, asynchronous requests
MAX_CONCURRENT_REQUESTS("bulk.max_concurrent_requests", Integer.class, Runtime.getRuntime().availableProcessors() - 1),
MAX_VOLUME_PER_REQUEST("max_volume_per_request"),
MAX_VOLUME_PER_REQUEST("bulk.max_volume_per_request", String.class, "1mb"),
FLUSH_INTERVAL("flush_interval");
FLUSH_INTERVAL("bulk.flush_interval", String.class, "30s");
boolean flag;
private final String name;
int num;
private final Class<?> type;
String string;
private final Object value;
Parameters(boolean flag) {
this.flag = flag;
Parameters(String name, Class<?> type, Object value) {
this.name = name;
this.type = type;
this.value = value;
}
Parameters(int num) {
this.num = num;
public String getName() {
return name;
}
Parameters(String string) {
this.string = string;
public Class<?> getType() {
return type;
}
boolean getValue() {
return flag;
public Boolean getBoolean() {
return type == Boolean.class ? (Boolean) value : Boolean.FALSE;
}
int getNum() {
return num;
public Integer getInteger() {
return type == Integer.class ? (Integer) value : 0;
}
String getString() {
return string;
public String getString() {
return type == String.class ? (String) value : null;
}
}

25
elx-common/src/main/java/org/xbib/elx/common/io/ClasspathURLStreamHandler.java
Unescape Escape View File

@ -1,25 +0,0 @@
package org.xbib.elx.common.io;
import java.io.IOException;
import java.net.URL;
import java.net.URLConnection;
import java.net.URLStreamHandler;
public class ClasspathURLStreamHandler extends URLStreamHandler {
private final ClassLoader classLoader;
public ClasspathURLStreamHandler() {
this.classLoader = getClass().getClassLoader();
}
public ClasspathURLStreamHandler(ClassLoader classLoader) {
this.classLoader = classLoader;
}
@Override
protected URLConnection openConnection(URL u) throws IOException {
final URL resourceUrl = classLoader.getResource(u.getPath());
return resourceUrl != null ? resourceUrl.openConnection() : null;
}
}

12
elx-common/src/main/java/org/xbib/elx/common/io/ClasspathURLStreamHandlerFactory.java
Unescape Escape View File

@ -1,12 +0,0 @@
package org.xbib.elx.common.io;
import java.net.URLStreamHandler;
import java.net.URLStreamHandlerFactory;
public class ClasspathURLStreamHandlerFactory implements URLStreamHandlerFactory {
@Override
public URLStreamHandler createURLStreamHandler(String protocol) {
return "classpath".equals(protocol) ? new ClasspathURLStreamHandler() : null;
}
}

4
elx-common/src/main/java/org/xbib/elx/common/io/package-info.java
Unescape Escape View File

@ -1,4 +0,0 @@
/**
* I/O helpers for Elasticsearch client extensions.
*/
package org.xbib.elx.common.io;

4
elx-common/src/main/java/org/xbib/elx/common/package-info.java
Unescape Escape View File

@ -1,4 +0,0 @@
/**
* Common classes for Elasticsearch client extensions.
*/
package org.xbib.elx.common;

4
elx-common/src/main/java/org/xbib/elx/common/util/package-info.java
Unescape Escape View File

@ -1,4 +0,0 @@
/**
* Utilities for Elasticsearch client extensions.
*/
package org.xbib.elx.common.util;

1
elx-common/src/main/resources/META-INF/services/java.net.URLStreamHandlerFactory
Unescape Escape View File

@ -1 +0,0 @@
org.xbib.elx.common.io.ClasspathURLStreamHandlerFactory

4
elx-common/src/test/java/org/xbib/elx/common/test/package-info.java
Unescape Escape View File

@ -1,4 +0,0 @@
/**
*
*/
package org.xbib.elx.common.test;

70
elx-node/src/test/java/org/xbib/elx/node/test/BulkClientTest.java
Unescape Escape View File

@ -2,7 +2,6 @@ package org.xbib.elx.node.test;
import org.apache.logging.log4j.LogManager;
import org.apache.logging.log4j.Logger;
import org.elasticsearch.common.unit.TimeValue;
import org.elasticsearch.common.xcontent.XContentBuilder;
import org.elasticsearch.common.xcontent.json.JsonXContent;
import org.junit.jupiter.api.Test;
@ -45,14 +44,12 @@ class BulkClientTest {
try (NodeBulkClient bulkClient = ClientBuilder.builder(helper.client)
.setBulkClientProvider(NodeBulkClientProvider.class)
.put(helper.getNodeSettings())
.put(Parameters.MAX_ACTIONS_PER_REQUEST.name(), MAX_ACTIONS_PER_REQUEST)
.put(Parameters.FLUSH_INTERVAL.name(), TimeValue.timeValueSeconds(30))
.put(Parameters.MAX_ACTIONS_PER_REQUEST.getName(), MAX_ACTIONS_PER_REQUEST)
.put(Parameters.FLUSH_INTERVAL.getName(), "30s")
.build()) {
IndexDefinition indexDefinition = new DefaultIndexDefinition();
indexDefinition.setFullIndexName("test");
indexDefinition.setType("doc");
IndexDefinition indexDefinition = new DefaultIndexDefinition("test", "doc");
bulkClient.newIndex(indexDefinition);
bulkClient.index("test", "doc", "1", true, "{ \"name\" : \"Hello World\"}"); // single doc ingest
bulkClient.index(indexDefinition, "1", true, "{ \"name\" : \"Hello World\"}"); // single doc ingest
bulkClient.flush();
bulkClient.waitForResponses(30L, TimeUnit.SECONDS);
assertEquals(1, bulkClient.getBulkController().getBulkMetric().getSucceeded().getCount());
@ -68,11 +65,9 @@ class BulkClientTest {
try (NodeBulkClient bulkClient = ClientBuilder.builder(helper.client)
.setBulkClientProvider(NodeBulkClientProvider.class)
.put(helper.getNodeSettings())
.put(Parameters.FLUSH_INTERVAL.name(), TimeValue.timeValueSeconds(5))
.put(Parameters.FLUSH_INTERVAL.getName(), "5s")
.build()) {
IndexDefinition indexDefinition = new DefaultIndexDefinition();
indexDefinition.setFullIndexName("test");
indexDefinition.setType("doc");
IndexDefinition indexDefinition = new DefaultIndexDefinition("test", "doc");
bulkClient.newIndex(indexDefinition);
}
}
@ -87,6 +82,7 @@ class BulkClientTest {
.setBulkClientProvider(NodeBulkClientProvider.class)
.put(helper.getNodeSettings())
.build()) {
IndexDefinition indexDefinition = new DefaultIndexDefinition("test", "doc");
XContentBuilder builder = JsonXContent.contentBuilder()
.startObject()
.startObject("doc")
@ -97,13 +93,9 @@ class BulkClientTest {
.endObject()
.endObject()
.endObject();
IndexDefinition indexDefinition = new DefaultIndexDefinition();
indexDefinition.setIndex("test");
indexDefinition.setFullIndexName("test");
indexDefinition.setType("doc");
indexDefinition.setMappings(builder.string());
bulkClient.newIndex(indexDefinition);
assertTrue(adminClient.getMapping("test").containsKey("properties"));
assertTrue(adminClient.getMapping(indexDefinition).containsKey("properties"));
}
}
@ -113,26 +105,25 @@ class BulkClientTest {
try (NodeBulkClient bulkClient = ClientBuilder.builder(helper.client)
.setBulkClientProvider(NodeBulkClientProvider.class)
.put(helper.getNodeSettings())
.put(Parameters.MAX_ACTIONS_PER_REQUEST.name(), MAX_ACTIONS_PER_REQUEST)
.put(Parameters.FLUSH_INTERVAL.name(), TimeValue.timeValueSeconds(60))
.put(Parameters.MAX_ACTIONS_PER_REQUEST.getName(), MAX_ACTIONS_PER_REQUEST)
.put(Parameters.FLUSH_INTERVAL.getName(), "60s")
.build()) {
IndexDefinition indexDefinition = new DefaultIndexDefinition();
indexDefinition.setFullIndexName("test");
indexDefinition.setType("doc");
IndexDefinition indexDefinition = new DefaultIndexDefinition("test", "doc");
bulkClient.newIndex(indexDefinition);
bulkClient.startBulk(indexDefinition);
for (int i = 0; i < ACTIONS; i++) {
bulkClient.index("test", "doc", null, false,
bulkClient.index(indexDefinition, null, false,
"{ \"name\" : \"" + helper.randomString(32) + "\"}");
}
bulkClient.flush();
bulkClient.stopBulk(indexDefinition);
bulkClient.waitForResponses(30L, TimeUnit.SECONDS);
assertEquals(numactions, bulkClient.getBulkController().getBulkMetric().getSucceeded().getCount());
if (bulkClient.getBulkController().getLastBulkError() != null) {
logger.error("error", bulkClient.getBulkController().getLastBulkError());
}
assertNull(bulkClient.getBulkController().getLastBulkError());
bulkClient.refreshIndex("test");
assertEquals(numactions, bulkClient.getSearchableDocs("test"));
bulkClient.refreshIndex(indexDefinition);
assertEquals(numactions, bulkClient.getSearchableDocs(indexDefinition));
}
}
@ -145,29 +136,20 @@ class BulkClientTest {
try (NodeBulkClient bulkClient = ClientBuilder.builder(helper.client)
.setBulkClientProvider(NodeBulkClientProvider.class)
.put(helper.getNodeSettings())
.put(Parameters.MAX_CONCURRENT_REQUESTS.name(), maxthreads)
.put(Parameters.MAX_ACTIONS_PER_REQUEST.name(), maxActionsPerRequest)
.put(Parameters.FLUSH_INTERVAL.name(), TimeValue.timeValueSeconds(60))
.put(Parameters.MAX_CONCURRENT_REQUESTS.getName(), maxthreads)
.put(Parameters.MAX_ACTIONS_PER_REQUEST.getName(), maxActionsPerRequest)
.put(Parameters.FLUSH_INTERVAL.getName(), "60s")
.build()) {
IndexDefinition indexDefinition = new DefaultIndexDefinition();
indexDefinition.setFullIndexName("test");
indexDefinition.setType("doc");
XContentBuilder builder = JsonXContent.contentBuilder()
.startObject()
.startObject("index")
.field("number_of_shards", 1)
.field("number_of_replicas", 0)
.endObject()
.endObject();
indexDefinition.setSettings(builder.string());
IndexDefinition indexDefinition = new DefaultIndexDefinition("test", "doc");
indexDefinition.setStartBulkRefreshSeconds(0);
bulkClient.newIndex(indexDefinition);
bulkClient.startBulk("test", 0, 1000);
bulkClient.startBulk(indexDefinition);
ExecutorService executorService = Executors.newFixedThreadPool(maxthreads);
final CountDownLatch latch = new CountDownLatch(maxthreads);
for (int i = 0; i < maxthreads; i++) {
executorService.execute(() -> {
for (int i1 = 0; i1 < actions; i1++) {
bulkClient.index("test", "doc", null, false,
bulkClient.index(indexDefinition, null, false,
"{ \"name\" : \"" + helper.randomString(32) + "\"}");
}
latch.countDown();
@ -185,10 +167,10 @@ class BulkClientTest {
} else {
logger.warn("latch timeout");
}
bulkClient.stopBulk("test", 30L, TimeUnit.SECONDS);
bulkClient.stopBulk(indexDefinition);
assertEquals(maxthreads * actions, bulkClient.getBulkController().getBulkMetric().getSucceeded().getCount());
bulkClient.refreshIndex("test");
assertEquals(maxthreads * actions, bulkClient.getSearchableDocs("test"));
bulkClient.refreshIndex(indexDefinition);
assertEquals(maxthreads * actions, bulkClient.getSearchableDocs(indexDefinition));
if (bulkClient.getBulkController().getLastBulkError() != null) {
logger.error("error", bulkClient.getBulkController().getLastBulkError());
}

12
elx-node/src/test/java/org/xbib/elx/node/test/DuplicateIDTest.java
Unescape Escape View File

@ -38,20 +38,18 @@ class DuplicateIDTest {
try (NodeBulkClient bulkClient = ClientBuilder.builder(helper.client)
.setBulkClientProvider(NodeBulkClientProvider.class)
.put(helper.getNodeSettings())
.put(Parameters.MAX_ACTIONS_PER_REQUEST.name(), MAX_ACTIONS_PER_REQUEST)
.put(Parameters.MAX_ACTIONS_PER_REQUEST.getName(), MAX_ACTIONS_PER_REQUEST)
.build()) {
IndexDefinition indexDefinition = new DefaultIndexDefinition();
indexDefinition.setFullIndexName("test");
indexDefinition.setType("doc");
IndexDefinition indexDefinition = new DefaultIndexDefinition("test", "doc");
bulkClient.newIndex(indexDefinition);
for (int i = 0; i < ACTIONS; i++) {
bulkClient.index("test", "doc", helper.randomString(1), false,
bulkClient.index(indexDefinition, helper.randomString(1), false,
"{ \"name\" : \"" + helper.randomString(32) + "\"}");
}
bulkClient.flush();
bulkClient.waitForResponses(30L, TimeUnit.SECONDS);
bulkClient.refreshIndex("test");
assertTrue(bulkClient.getSearchableDocs("test") < ACTIONS);
bulkClient.refreshIndex(indexDefinition);
assertTrue(bulkClient.getSearchableDocs(indexDefinition) < ACTIONS);
assertEquals(numactions, bulkClient.getBulkController().getBulkMetric().getSucceeded().getCount());
if (bulkClient.getBulkController().getLastBulkError() != null) {
logger.error("error", bulkClient.getBulkController().getLastBulkError());

26
elx-node/src/test/java/org/xbib/elx/node/test/IndexPruneTest.java
Unescape Escape View File

@ -2,8 +2,6 @@ package org.xbib.elx.node.test;
import org.apache.logging.log4j.LogManager;
import org.apache.logging.log4j.Logger;
import org.elasticsearch.common.xcontent.XContentBuilder;
import org.elasticsearch.common.xcontent.json.JsonXContent;
import org.junit.jupiter.api.Test;
import org.junit.jupiter.api.extension.ExtendWith;
import org.xbib.elx.api.IndexDefinition;
@ -47,33 +45,23 @@ class IndexPruneTest {
.setBulkClientProvider(NodeBulkClientProvider.class)
.put(helper.getNodeSettings())
.build()) {
XContentBuilder builder = JsonXContent.contentBuilder()
.startObject()
.startObject("index")
.field("number_of_shards", 1)
.field("number_of_replicas", 0)
.endObject()
.endObject();
IndexDefinition indexDefinition = new DefaultIndexDefinition();
indexDefinition.setIndex("test_prune");
indexDefinition.setType("doc");
IndexDefinition indexDefinition = new DefaultIndexDefinition("test_prune", "doc");
indexDefinition.setFullIndexName("test_prune1");
indexDefinition.setSettings(builder.string());
bulkClient.newIndex(indexDefinition);
indexDefinition.setShift(true);
adminClient.shiftIndex(indexDefinition, Collections.emptyList());
adminClient.shiftIndex(indexDefinition, Collections.emptyList(), null);
indexDefinition.setFullIndexName("test_prune2");
bulkClient.newIndex(indexDefinition);
indexDefinition.setShift(true);
adminClient.shiftIndex(indexDefinition, Collections.emptyList());
adminClient.shiftIndex(indexDefinition, Collections.emptyList(), null);
indexDefinition.setFullIndexName("test_prune3");
bulkClient.newIndex(indexDefinition);
indexDefinition.setShift(true);
adminClient.shiftIndex(indexDefinition, Collections.emptyList());
adminClient.shiftIndex(indexDefinition, Collections.emptyList(), null);
indexDefinition.setFullIndexName("test_prune4");
bulkClient.newIndex(indexDefinition);
indexDefinition.setShift(true);
adminClient.shiftIndex(indexDefinition, Collections.emptyList());
adminClient.shiftIndex(indexDefinition, Collections.emptyList(), null);
IndexRetention indexRetention = new DefaultIndexRetention();
indexRetention.setDelta(2);
indexRetention.setMinToKeep(2);
@ -90,7 +78,9 @@ class IndexPruneTest {
assertFalse(indexPruneResult.getDeletedIndices().contains("test_prune4"));
List<Boolean> list = new ArrayList<>();
for (String index : Arrays.asList("test_prune1", "test_prune2", "test_prune3", "test_prune4")) {
list.add(adminClient.isIndexExists(index));
IndexDefinition indexDefinition1 = new DefaultIndexDefinition(index, null);
indexDefinition1.setFullIndexName(index);
list.add(adminClient.isIndexExists(indexDefinition1));
}
logger.info(list);
assertFalse(list.get(0));

24
elx-node/src/test/java/org/xbib/elx/node/test/IndexShiftTest.java
Unescape Escape View File

@ -4,8 +4,6 @@ import org.apache.logging.log4j.LogManager;
import org.apache.logging.log4j.Logger;
import org.elasticsearch.action.admin.indices.alias.IndicesAliasesRequest;
import org.elasticsearch.cluster.metadata.AliasAction;
import org.elasticsearch.common.xcontent.XContentBuilder;
import org.elasticsearch.common.xcontent.json.JsonXContent;
import org.elasticsearch.index.query.QueryBuilders;
import org.junit.jupiter.api.Test;
import org.junit.jupiter.api.extension.ExtendWith;
@ -46,27 +44,18 @@ class IndexShiftTest {
.setBulkClientProvider(NodeBulkClientProvider.class)
.put(helper.getNodeSettings())
.build()) {
XContentBuilder builder = JsonXContent.contentBuilder()
.startObject()
.startObject("index")
.field("number_of_shards", 1)
.field("number_of_replicas", 0)
.endObject()
.endObject();
IndexDefinition indexDefinition = new DefaultIndexDefinition();
indexDefinition.setIndex("test");
indexDefinition.setType("doc");
IndexDefinition indexDefinition = new DefaultIndexDefinition("test", "doc");
indexDefinition.setFullIndexName("test_shift");
indexDefinition.setSettings(builder.string());
bulkClient.newIndex(indexDefinition);
for (int i = 0; i < 1; i++) {
bulkClient.index("test_shift", "doc", helper.randomString(1), false,
bulkClient.index(indexDefinition, helper.randomString(1), false,
"{ \"name\" : \"" + helper.randomString(32) + "\"}");
}
bulkClient.flush();
bulkClient.waitForResponses(30L, TimeUnit.SECONDS);
indexDefinition.setShift(true);
IndexShiftResult indexShiftResult = adminClient.shiftIndex(indexDefinition, Arrays.asList("a", "b", "c"));
IndexShiftResult indexShiftResult =
adminClient.shiftIndex(indexDefinition, Arrays.asList("a", "b", "c"), null);
assertTrue(indexShiftResult.getNewAliases().contains("a"));
assertTrue(indexShiftResult.getNewAliases().contains("b"));
assertTrue(indexShiftResult.getNewAliases().contains("c"));
@ -77,7 +66,8 @@ class IndexShiftTest {
assertTrue(aliases.containsKey("c"));
assertTrue(aliases.containsKey("test"));
Optional<String> resolved = adminClient.resolveAlias("test").stream().findFirst();
aliases = resolved.isPresent() ? adminClient.getAliases(resolved.get()) : Collections.emptyMap();
aliases = resolved.isPresent() ?
adminClient.getAliases(resolved.get()) : Collections.emptyMap();
assertTrue(aliases.containsKey("a"));
assertTrue(aliases.containsKey("b"));
assertTrue(aliases.containsKey("c"));
@ -85,7 +75,7 @@ class IndexShiftTest {
indexDefinition.setFullIndexName("test_shift2");
bulkClient.newIndex(indexDefinition);
for (int i = 0; i < 1; i++) {
bulkClient.index("test_shift2", "doc", helper.randomString(1), false,
bulkClient.index(indexDefinition, helper.randomString(1), false,
"{ \"name\" : \"" + helper.randomString(32) + "\"}");
}
bulkClient.flush();

28
elx-node/src/test/java/org/xbib/elx/node/test/SearchTest.java
Unescape Escape View File

@ -2,6 +2,7 @@ package org.xbib.elx.node.test;
import static org.junit.jupiter.api.Assertions.assertEquals;
import static org.junit.jupiter.api.Assertions.assertNull;
import static org.junit.jupiter.api.Assertions.assertTrue;
import org.apache.logging.log4j.LogManager;
import org.apache.logging.log4j.Logger;
import org.elasticsearch.common.unit.TimeValue;
@ -26,9 +27,9 @@ class SearchTest {
private static final Logger logger = LogManager.getLogger(SearchTest.class.getName());
private static final Long ACTIONS = 1000L;
private static final Long ACTIONS = 100000L;
private static final Long MAX_ACTIONS_PER_REQUEST = 100L;
private static final Long MAX_ACTIONS_PER_REQUEST = 1000L;
private final TestExtension.Helper helper;
@ -39,23 +40,22 @@ class SearchTest {
@Test
void testDocStream() throws Exception {
long numactions = ACTIONS;
IndexDefinition indexDefinition = new DefaultIndexDefinition("test", "doc");
try (NodeBulkClient bulkClient = ClientBuilder.builder(helper.client)
.setBulkClientProvider(NodeBulkClientProvider.class)
.put(helper.getNodeSettings())
.put(Parameters.MAX_ACTIONS_PER_REQUEST.name(), MAX_ACTIONS_PER_REQUEST)
.put(Parameters.MAX_ACTIONS_PER_REQUEST.getName(), MAX_ACTIONS_PER_REQUEST)
.build()) {
IndexDefinition indexDefinition = new DefaultIndexDefinition();
indexDefinition.setFullIndexName("test");
indexDefinition.setType("doc");
bulkClient.newIndex(indexDefinition);
bulkClient.startBulk(indexDefinition);
for (int i = 0; i < numactions; i++) {
bulkClient.index("test", "doc", null, false,
bulkClient.index(indexDefinition, null, false,
"{ \"name\" : \"" + helper.randomString(32) + "\"}");
}
bulkClient.flush();
bulkClient.waitForResponses(30L, TimeUnit.SECONDS);
bulkClient.refreshIndex("test");
assertEquals(numactions, bulkClient.getSearchableDocs("test"));
bulkClient.stopBulk(indexDefinition);
assertTrue(bulkClient.waitForResponses(30L, TimeUnit.SECONDS));
bulkClient.refreshIndex(indexDefinition);
assertEquals(numactions, bulkClient.getSearchableDocs(indexDefinition));
assertEquals(numactions, bulkClient.getBulkController().getBulkMetric().getSucceeded().getCount());
if (bulkClient.getBulkController().getLastBulkError() != null) {
logger.error("error", bulkClient.getBulkController().getLastBulkError());
@ -68,7 +68,7 @@ class SearchTest {
.build()) {
// test stream count
Stream<SearchHit> stream = searchClient.search(qb -> qb
.setIndices("test")
.setIndices(indexDefinition.getFullIndexName())
.setQuery(QueryBuilders.matchAllQuery()),
TimeValue.timeValueMillis(100), 570);
long count = stream.count();
@ -78,7 +78,7 @@ class SearchTest {
assertEquals(1L, searchClient.getSearchMetric().getEmptyQueries().getCount());
// test stream docs
stream = searchClient.search(qb -> qb
.setIndices("test")
.setIndices(indexDefinition.getFullIndexName())
.setQuery(QueryBuilders.matchAllQuery()),
TimeValue.timeValueMillis(10), 79);
final AtomicInteger hitcount = new AtomicInteger();
@ -89,7 +89,7 @@ class SearchTest {
assertEquals(2L, searchClient.getSearchMetric().getEmptyQueries().getCount());
// test stream doc ids
Stream<String> ids = searchClient.getIds(qb -> qb
.setIndices("test")
.setIndices(indexDefinition.getFullIndexName())
.setQuery(QueryBuilders.matchAllQuery()));
final AtomicInteger idcount = new AtomicInteger();
ids.forEach(id -> {

31
elx-node/src/test/java/org/xbib/elx/node/test/SmokeTest.java
Unescape Escape View File

@ -7,6 +7,7 @@ import org.junit.jupiter.api.Test;
import org.junit.jupiter.api.extension.ExtendWith;
import org.xbib.elx.common.ClientBuilder;
import org.xbib.elx.api.IndexDefinition;
import org.xbib.elx.common.DefaultIndexDefinition;
import org.xbib.elx.node.NodeAdminClient;
import org.xbib.elx.node.NodeAdminClientProvider;
import org.xbib.elx.node.NodeBulkClient;
@ -14,6 +15,7 @@ import org.xbib.elx.node.NodeBulkClientProvider;
import java.util.concurrent.TimeUnit;
import static org.junit.jupiter.api.Assertions.assertTrue;
import static org.junit.jupiter.api.Assertions.assertEquals;
import static org.junit.jupiter.api.Assertions.assertNull;
@ -38,34 +40,29 @@ class SmokeTest {
.setBulkClientProvider(NodeBulkClientProvider.class)
.put(helper.getNodeSettings())
.build()) {
assertEquals(helper.getClusterName(), adminClient.getClusterName());
IndexDefinition indexDefinition =
adminClient.buildIndexDefinitionFromSettings("test_smoke_definition", Settings.EMPTY);
new DefaultIndexDefinition(adminClient, "test_smoke", "doc", Settings.EMPTY);
assertEquals("test_smoke", indexDefinition.getIndex());
assertTrue(indexDefinition.getFullIndexName().startsWith("test_smoke"));
assertEquals(0, indexDefinition.getReplicaLevel());
assertEquals(helper.getClusterName(), adminClient.getClusterName());
indexDefinition.setFullIndexName("test_smoke");
indexDefinition.setType("doc");
bulkClient.newIndex(indexDefinition);
logger.info("new index: done");
bulkClient.index("test_smoke", "doc", "1", true, "{ \"name\" : \"Hello World\"}"); // single doc ingest
logger.info("index doc: done");
bulkClient.index(indexDefinition, "1", true, "{ \"name\" : \"Hello World\"}"); // single doc ingest
bulkClient.flush();
logger.info("flush: done");
bulkClient.waitForResponses(30, TimeUnit.SECONDS);
logger.info("wait: done");
adminClient.checkMapping("test_smoke");
logger.info("check mapping: done");
bulkClient.update("test_smoke", "doc", "1", "{ \"name\" : \"Another name\"}");
bulkClient.delete("test_smoke", "doc", "1");
adminClient.checkMapping(indexDefinition);
bulkClient.update(indexDefinition, "1", "{ \"name\" : \"Another name\"}");
bulkClient.delete(indexDefinition, "1");
bulkClient.flush();
bulkClient.waitForResponses(30, TimeUnit.SECONDS);
bulkClient.index("test_smoke", "doc", "1", true, "{ \"name\" : \"Hello World\"}");
bulkClient.delete("test_smoke", "doc", "1");
bulkClient.index(indexDefinition, "1", true, "{ \"name\" : \"Hello World\"}");
bulkClient.delete(indexDefinition, "1");
bulkClient.flush();
bulkClient.waitForResponses(30, TimeUnit.SECONDS);
adminClient.deleteIndex("test_smoke");
logger.info("delete index: done");
adminClient.deleteIndex(indexDefinition);
bulkClient.newIndex(indexDefinition);
bulkClient.index(indexDefinition.getFullIndexName(), "doc", "1", true, "{ \"name\" : \"Hello World\"}");
bulkClient.index(indexDefinition, "1", true, "{ \"name\" : \"Hello World\"}");
bulkClient.flush();
bulkClient.waitForResponses(30, TimeUnit.SECONDS);
adminClient.updateReplicaLevel(indexDefinition, 2);

4
elx-node/src/test/java/org/xbib/elx/node/test/package-info.java
Unescape Escape View File

@ -1,4 +0,0 @@
/**
*
*/
package org.xbib.elx.node.test;

2
elx-node/src/test/resources/log4j2-test.xml
Unescape Escape View File

@ -6,7 +6,7 @@
</Console>
</appenders>
<Loggers>
<Root level="info">
<Root level="debug">
<AppenderRef ref="Console" />
</Root>
</Loggers>

2
elx-common/src/main/java/org/xbib/elx/common/util/NetworkUtils.java → elx-transport/src/main/java/org/xbib/elx/transport/NetworkUtils.java
Unescape Escape View File

@ -1,4 +1,4 @@
package org.xbib.elx.common.util;
package org.xbib.elx.transport;
import org.apache.logging.log4j.LogManager;
import org.apache.logging.log4j.Logger;

1
elx-transport/src/main/java/org/xbib/elx/transport/TransportClientHelper.java
Unescape Escape View File

@ -16,7 +16,6 @@ import org.elasticsearch.common.settings.Settings;
import org.elasticsearch.common.transport.InetSocketTransportAddress;
import org.elasticsearch.common.transport.TransportAddress;
import org.jboss.netty.channel.DefaultChannelFuture;
import org.xbib.elx.common.util.NetworkUtils;
import java.io.IOException;
import java.net.InetAddress;
import java.util.ArrayList;

4
elx-transport/src/main/java/org/xbib/elx/transport/package-info.java
Unescape Escape View File

@ -1,4 +0,0 @@
/**
* Classes for Elasticsearch transport client extensions.
*/
package org.xbib.elx.transport;

67
elx-transport/src/test/java/org/xbib/elx/transport/test/BulkClientTest.java
Unescape Escape View File

@ -2,7 +2,6 @@ package org.xbib.elx.transport.test;
import org.apache.logging.log4j.LogManager;
import org.apache.logging.log4j.Logger;
import org.elasticsearch.common.unit.TimeValue;
import org.elasticsearch.common.xcontent.XContentBuilder;
import org.elasticsearch.common.xcontent.json.JsonXContent;
import org.junit.jupiter.api.Test;
@ -45,14 +44,12 @@ class BulkClientTest {
try (TransportBulkClient bulkClient = ClientBuilder.builder()
.setBulkClientProvider(TransportBulkClientProvider.class)
.put(helper.getTransportSettings())
.put(Parameters.MAX_ACTIONS_PER_REQUEST.name(), MAX_ACTIONS_PER_REQUEST)
.put(Parameters.FLUSH_INTERVAL.name(), TimeValue.timeValueSeconds(30))
.put(Parameters.MAX_ACTIONS_PER_REQUEST.getName(), MAX_ACTIONS_PER_REQUEST)
.put(Parameters.FLUSH_INTERVAL.getName(), "30s")
.build()) {
IndexDefinition indexDefinition = new DefaultIndexDefinition();
indexDefinition.setFullIndexName("test");
indexDefinition.setType("doc");
IndexDefinition indexDefinition = new DefaultIndexDefinition("test", "doc");
bulkClient.newIndex(indexDefinition);
bulkClient.index("test", "docd", "1", true, "{ \"name\" : \"Hello World\"}"); // single doc ingest
bulkClient.index(indexDefinition, "1", true, "{ \"name\" : \"Hello World\"}"); // single doc ingest
bulkClient.flush();
bulkClient.waitForResponses(30L, TimeUnit.SECONDS);
assertEquals(1, bulkClient.getBulkController().getBulkMetric().getSucceeded().getCount());
@ -69,9 +66,7 @@ class BulkClientTest {
.setBulkClientProvider(TransportBulkClientProvider.class)
.put(helper.getTransportSettings())
.build()) {
IndexDefinition indexDefinition = new DefaultIndexDefinition();
indexDefinition.setFullIndexName("test");
indexDefinition.setType("doc");
IndexDefinition indexDefinition = new DefaultIndexDefinition("test", "doc");
bulkClient.newIndex(indexDefinition);
}
}
@ -96,12 +91,10 @@ class BulkClientTest {
.endObject()
.endObject()
.endObject();
IndexDefinition indexDefinition = new DefaultIndexDefinition();
indexDefinition.setFullIndexName("test");
indexDefinition.setType("doc");
IndexDefinition indexDefinition = new DefaultIndexDefinition("test", "doc");
indexDefinition.setMappings(builder.string());
bulkClient.newIndex(indexDefinition);
assertTrue(adminClient.getMapping("test").containsKey("properties"));
assertTrue(adminClient.getMapping(indexDefinition).containsKey("properties"));
}
}
@ -111,15 +104,13 @@ class BulkClientTest {
try (TransportBulkClient bulkClient = ClientBuilder.builder()
.setBulkClientProvider(TransportBulkClientProvider.class)
.put(helper.getTransportSettings())
.put(Parameters.MAX_ACTIONS_PER_REQUEST.name(), MAX_ACTIONS_PER_REQUEST)
.put(Parameters.FLUSH_INTERVAL.name(), TimeValue.timeValueSeconds(60))
.put(Parameters.MAX_ACTIONS_PER_REQUEST.getName(), MAX_ACTIONS_PER_REQUEST)
.put(Parameters.FLUSH_INTERVAL.getName(), "60s")
.build()) {
IndexDefinition indexDefinition = new DefaultIndexDefinition();
indexDefinition.setFullIndexName("test");
indexDefinition.setType("doc");
IndexDefinition indexDefinition = new DefaultIndexDefinition("test", "doc");
bulkClient.newIndex(indexDefinition);
for (int i = 0; i < ACTIONS; i++) {
bulkClient.index("test", "docs", null, false,
bulkClient.index(indexDefinition, null, false,
"{ \"name\" : \"" + helper.randomString(32) + "\"}");
}
bulkClient.flush();
@ -129,8 +120,8 @@ class BulkClientTest {
logger.error("error", bulkClient.getBulkController().getLastBulkError());
}
assertNull(bulkClient.getBulkController().getLastBulkError());
bulkClient.refreshIndex("test");
assertEquals(numactions, bulkClient.getSearchableDocs("test"));
bulkClient.refreshIndex(indexDefinition);
assertEquals(numactions, bulkClient.getSearchableDocs(indexDefinition));
}
}
@ -143,32 +134,24 @@ class BulkClientTest {
try (TransportBulkClient bulkClient = ClientBuilder.builder()
.setBulkClientProvider(TransportBulkClientProvider.class)
.put(helper.getTransportSettings())
.put(Parameters.MAX_CONCURRENT_REQUESTS.name(), maxthreads * 2)
.put(Parameters.MAX_ACTIONS_PER_REQUEST.name(), maxActionsPerRequest)
.put(Parameters.FLUSH_INTERVAL.name(), TimeValue.timeValueSeconds(60))
.put(Parameters.ENABLE_BULK_LOGGING.name(), "true")
.put(Parameters.MAX_CONCURRENT_REQUESTS.getName(), maxthreads * 2)
.put(Parameters.MAX_ACTIONS_PER_REQUEST.getName(), maxActionsPerRequest)
.put(Parameters.FLUSH_INTERVAL.getName(), "60s")
.put(Parameters.ENABLE_BULK_LOGGING.getName(), Boolean.TRUE)
.build()) {
XContentBuilder builder = JsonXContent.contentBuilder()
.startObject()
.startObject("index")
.field("number_of_shards", 1)
.field("number_of_replicas", 0)
.endObject()
.endObject();
IndexDefinition indexDefinition = new DefaultIndexDefinition();
indexDefinition.setIndex("test");
indexDefinition.setType("doc");
IndexDefinition indexDefinition = new DefaultIndexDefinition("test", "doc");
indexDefinition.setFullIndexName("test");
indexDefinition.setSettings(builder.string());
indexDefinition.setStartBulkRefreshSeconds(0);
indexDefinition.setStopBulkRefreshSeconds(60);
bulkClient.newIndex(indexDefinition);
bulkClient.startBulk("test", 0, 1000);
bulkClient.startBulk(indexDefinition);
logger.info("index created");
ExecutorService executorService = Executors.newFixedThreadPool(maxthreads);
final CountDownLatch latch = new CountDownLatch(maxthreads);
for (int i = 0; i < maxthreads; i++) {
executorService.execute(() -> {
for (int i1 = 0; i1 < actions; i1++) {
bulkClient.index("test", "docs", null, false,
bulkClient.index(indexDefinition, null, false,
"{ \"name\" : \"" + helper.randomString(32) + "\"}");
}
latch.countDown();
@ -186,10 +169,10 @@ class BulkClientTest {
} else {
logger.warn("latch timeout");
}
bulkClient.stopBulk("test", 30L, TimeUnit.SECONDS);
bulkClient.stopBulk(indexDefinition);
assertEquals(maxthreads * actions, bulkClient.getBulkController().getBulkMetric().getSucceeded().getCount());
bulkClient.refreshIndex("test");
assertEquals(maxthreads * actions, bulkClient.getSearchableDocs("test"));
bulkClient.refreshIndex(indexDefinition);
assertEquals(maxthreads * actions, bulkClient.getSearchableDocs(indexDefinition));
if (bulkClient.getBulkController().getLastBulkError() != null) {
logger.error("error", bulkClient.getBulkController().getLastBulkError());
}

12
elx-transport/src/test/java/org/xbib/elx/transport/test/DuplicateIDTest.java
Unescape Escape View File

@ -36,20 +36,18 @@ class DuplicateIDTest {
try (TransportBulkClient bulkClient = ClientBuilder.builder()
.setBulkClientProvider(TransportBulkClientProvider.class)
.put(helper.getTransportSettings())
.put(Parameters.MAX_ACTIONS_PER_REQUEST.name(), MAX_ACTIONS_PER_REQUEST)
.put(Parameters.MAX_ACTIONS_PER_REQUEST.getName(), MAX_ACTIONS_PER_REQUEST)
.build()) {
IndexDefinition indexDefinition = new DefaultIndexDefinition();
indexDefinition.setFullIndexName("test");
indexDefinition.setType("doc");
IndexDefinition indexDefinition = new DefaultIndexDefinition("test", "doc");
bulkClient.newIndex(indexDefinition);
for (int i = 0; i < ACTIONS; i++) {
bulkClient.index("test", "docs", helper.randomString(1), false,
bulkClient.index(indexDefinition, helper.randomString(1), false,
"{ \"name\" : \"" + helper.randomString(32) + "\"}");
}
bulkClient.flush();
bulkClient.waitForResponses(30L, TimeUnit.SECONDS);
bulkClient.refreshIndex("test");
assertTrue(bulkClient.getSearchableDocs("test") < ACTIONS);
bulkClient.refreshIndex(indexDefinition);
assertTrue(bulkClient.getSearchableDocs(indexDefinition) < ACTIONS);
assertEquals(numactions, bulkClient.getBulkController().getBulkMetric().getSucceeded().getCount());
if (bulkClient.getBulkController().getLastBulkError() != null) {
logger.error("error", bulkClient.getBulkController().getLastBulkError());

26
elx-transport/src/test/java/org/xbib/elx/transport/test/IndexPruneTest.java
Unescape Escape View File

@ -2,8 +2,6 @@ package org.xbib.elx.transport.test;
import org.apache.logging.log4j.LogManager;
import org.apache.logging.log4j.Logger;
import org.elasticsearch.common.xcontent.XContentBuilder;
import org.elasticsearch.common.xcontent.json.JsonXContent;
import org.junit.jupiter.api.Test;
import org.junit.jupiter.api.extension.ExtendWith;
import org.xbib.elx.api.IndexDefinition;
@ -48,33 +46,23 @@ class IndexPruneTest {
.setBulkClientProvider(TransportBulkClientProvider.class)
.put(helper.getTransportSettings())
.build()) {
XContentBuilder builder = JsonXContent.contentBuilder()
.startObject()
.startObject("index")
.field("number_of_shards", 1)
.field("number_of_replicas", 0)
.endObject()
.endObject();
IndexDefinition indexDefinition = new DefaultIndexDefinition();
indexDefinition.setIndex("test_prune");
indexDefinition.setType("doc");
IndexDefinition indexDefinition = new DefaultIndexDefinition("test", "doc");
indexDefinition.setFullIndexName("test_prune1");
indexDefinition.setSettings(builder.string());
bulkClient.newIndex(indexDefinition);
indexDefinition.setShift(true);
adminClient.shiftIndex(indexDefinition, Collections.emptyList());
adminClient.shiftIndex(indexDefinition, Collections.emptyList(), null);
indexDefinition.setFullIndexName("test_prune2");
bulkClient.newIndex(indexDefinition);
indexDefinition.setShift(true);
adminClient.shiftIndex(indexDefinition, Collections.emptyList());
adminClient.shiftIndex(indexDefinition, Collections.emptyList(), null);
indexDefinition.setFullIndexName("test_prune3");
bulkClient.newIndex(indexDefinition);
indexDefinition.setShift(true);
adminClient.shiftIndex(indexDefinition, Collections.emptyList());
adminClient.shiftIndex(indexDefinition, Collections.emptyList(), null);
indexDefinition.setFullIndexName("test_prune4");
bulkClient.newIndex(indexDefinition);
indexDefinition.setShift(true);
adminClient.shiftIndex(indexDefinition, Collections.emptyList());
adminClient.shiftIndex(indexDefinition, Collections.emptyList(), null);
IndexRetention indexRetention = new DefaultIndexRetention();
indexRetention.setDelta(2);
indexRetention.setMinToKeep(2);
@ -91,7 +79,9 @@ class IndexPruneTest {
assertFalse(indexPruneResult.getDeletedIndices().contains("test_prune4"));
List<Boolean> list = new ArrayList<>();
for (String index : Arrays.asList("test_prune1", "test_prune2", "test_prune3", "test_prune4")) {
list.add(adminClient.isIndexExists(index));
IndexDefinition indexDefinition1 = new DefaultIndexDefinition(index, null);
indexDefinition1.setFullIndexName(index);
list.add(adminClient.isIndexExists(indexDefinition1));
}
logger.info(list);
assertFalse(list.get(0));

23
elx-transport/src/test/java/org/xbib/elx/transport/test/IndexShiftTest.java
Unescape Escape View File

@ -4,8 +4,6 @@ import org.apache.logging.log4j.LogManager;
import org.apache.logging.log4j.Logger;
import org.elasticsearch.action.admin.indices.alias.IndicesAliasesRequest;
import org.elasticsearch.cluster.metadata.AliasAction;
import org.elasticsearch.common.xcontent.XContentBuilder;
import org.elasticsearch.common.xcontent.json.JsonXContent;
import org.elasticsearch.index.query.QueryBuilders;
import org.junit.jupiter.api.Test;
import org.junit.jupiter.api.extension.ExtendWith;
@ -48,28 +46,18 @@ class IndexShiftTest {
.setBulkClientProvider(TransportBulkClientProvider.class)
.put(helper.getTransportSettings())
.build()) {
XContentBuilder builder = JsonXContent.contentBuilder()
.startObject()
.startObject("index")
.field("number_of_shards", 1)
.field("number_of_replicas", 0)
.endObject()
.endObject();
IndexDefinition indexDefinition = new DefaultIndexDefinition();
indexDefinition.setIndex("test");
indexDefinition.setType("doc");
IndexDefinition indexDefinition = new DefaultIndexDefinition("test", "doc");
indexDefinition.setFullIndexName("test_shift");
indexDefinition.setSettings(builder.string());
bulkClient.newIndex(indexDefinition);
for (int i = 0; i < 1; i++) {
bulkClient.index("test_shift", "doc", helper.randomString(1), false,
bulkClient.index(indexDefinition, helper.randomString(1), false,
"{ \"name\" : \"" + helper.randomString(32) + "\"}");
}
bulkClient.flush();
bulkClient.waitForResponses(30L, TimeUnit.SECONDS);
indexDefinition.setShift(true);
IndexShiftResult indexShiftResult =
adminClient.shiftIndex(indexDefinition, Arrays.asList("a", "b", "c"));
adminClient.shiftIndex(indexDefinition, Arrays.asList("a", "b", "c"), null);
assertTrue(indexShiftResult.getNewAliases().contains("a"));
assertTrue(indexShiftResult.getNewAliases().contains("b"));
assertTrue(indexShiftResult.getNewAliases().contains("c"));
@ -80,7 +68,8 @@ class IndexShiftTest {
assertTrue(aliases.containsKey("c"));
assertTrue(aliases.containsKey("test"));
Optional<String> resolved = adminClient.resolveAlias("test").stream().findFirst();
aliases = resolved.isPresent() ? adminClient.getAliases(resolved.get()) : Collections.emptyMap();
aliases = resolved.isPresent() ?
adminClient.getAliases(resolved.get()) : Collections.emptyMap();
assertTrue(aliases.containsKey("a"));
assertTrue(aliases.containsKey("b"));
assertTrue(aliases.containsKey("c"));
@ -88,7 +77,7 @@ class IndexShiftTest {
indexDefinition.setFullIndexName("test_shift2");
bulkClient.newIndex(indexDefinition);
for (int i = 0; i < 1; i++) {
bulkClient.index("test_shift2", "doc", helper.randomString(1), false,
bulkClient.index(indexDefinition, helper.randomString(1), false,
"{ \"name\" : \"" + helper.randomString(32) + "\"}");
}
bulkClient.flush();

26
elx-transport/src/test/java/org/xbib/elx/transport/test/SearchTest.java
Unescape Escape View File

@ -22,13 +22,14 @@ import java.util.stream.Stream;
import static org.junit.jupiter.api.Assertions.assertEquals;
import static org.junit.jupiter.api.Assertions.assertNull;
import static org.junit.jupiter.api.Assertions.assertTrue;
@ExtendWith(TestExtension.class)
class SearchTest {
private static final Logger logger = LogManager.getLogger(SearchTest.class.getName());
private static final Long ACTIONS = 1000L;
private static final Long ACTIONS = 100000L;
private static final Long MAX_ACTIONS_PER_REQUEST = 100L;
@ -41,23 +42,22 @@ class SearchTest {
@Test
void testDocStream() throws Exception {
long numactions = ACTIONS;
IndexDefinition indexDefinition = new DefaultIndexDefinition("test", "doc");
try (TransportBulkClient bulkClient = ClientBuilder.builder()
.setBulkClientProvider(TransportBulkClientProvider.class)
.put(helper.getTransportSettings())
.put(Parameters.MAX_ACTIONS_PER_REQUEST.name(), MAX_ACTIONS_PER_REQUEST)
.put(Parameters.MAX_ACTIONS_PER_REQUEST.getName(), MAX_ACTIONS_PER_REQUEST)
.build()) {
IndexDefinition indexDefinition = new DefaultIndexDefinition();
indexDefinition.setFullIndexName("test");
indexDefinition.setType("doc");
bulkClient.newIndex(indexDefinition);
bulkClient.startBulk(indexDefinition);
for (int i = 0; i < numactions; i++) {
bulkClient.index("test", "doc", null, false,
bulkClient.index(indexDefinition, null, false,
"{ \"name\" : \"" + helper.randomString(32) + "\"}");
}
bulkClient.flush();
bulkClient.waitForResponses(30L, TimeUnit.SECONDS);
bulkClient.refreshIndex("test");
assertEquals(numactions, bulkClient.getSearchableDocs("test"));
bulkClient.stopBulk(indexDefinition);
assertTrue(bulkClient.waitForResponses(30L, TimeUnit.SECONDS));
bulkClient.refreshIndex(indexDefinition);
assertEquals(numactions, bulkClient.getSearchableDocs(indexDefinition));
assertEquals(numactions, bulkClient.getBulkController().getBulkMetric().getSucceeded().getCount());
if (bulkClient.getBulkController().getLastBulkError() != null) {
logger.error("error", bulkClient.getBulkController().getLastBulkError());
@ -70,7 +70,7 @@ class SearchTest {
.build()) {
// test stream count
Stream<SearchHit> stream = searchClient.search(qb -> qb
.setIndices("test")
.setIndices(indexDefinition.getFullIndexName())
.setQuery(QueryBuilders.matchAllQuery()),
TimeValue.timeValueMillis(100), 570);
long count = stream.count();
@ -80,7 +80,7 @@ class SearchTest {
assertEquals(1L, searchClient.getSearchMetric().getEmptyQueries().getCount());
// test stream docs
stream = searchClient.search(qb -> qb
.setIndices("test")
.setIndices(indexDefinition.getFullIndexName())
.setQuery(QueryBuilders.matchAllQuery()),
TimeValue.timeValueMillis(10), 79);
final AtomicInteger hitcount = new AtomicInteger();
@ -91,7 +91,7 @@ class SearchTest {
assertEquals(2L, searchClient.getSearchMetric().getEmptyQueries().getCount());
// test stream doc ids
Stream<String> ids = searchClient.getIds(qb -> qb
.setIndices("test")
.setIndices(indexDefinition.getFullIndexName())
.setQuery(QueryBuilders.matchAllQuery()));
final AtomicInteger idcount = new AtomicInteger();
ids.forEach(id -> {

28
elx-transport/src/test/java/org/xbib/elx/transport/test/SmokeTest.java
Unescape Escape View File

@ -7,6 +7,7 @@ import org.junit.jupiter.api.Test;
import org.junit.jupiter.api.extension.ExtendWith;
import org.xbib.elx.api.IndexDefinition;
import org.xbib.elx.common.ClientBuilder;
import org.xbib.elx.common.DefaultIndexDefinition;
import org.xbib.elx.transport.TransportAdminClient;
import org.xbib.elx.transport.TransportAdminClientProvider;
import org.xbib.elx.transport.TransportBulkClient;
@ -16,6 +17,7 @@ import java.util.concurrent.TimeUnit;
import static org.junit.jupiter.api.Assertions.assertEquals;
import static org.junit.jupiter.api.Assertions.assertNull;
import static org.junit.jupiter.api.Assertions.assertTrue;
@ExtendWith(TestExtension.class)
class SmokeTest {
@ -39,28 +41,24 @@ class SmokeTest {
.put(helper.getTransportSettings())
.build()) {
IndexDefinition indexDefinition =
adminClient.buildIndexDefinitionFromSettings("test_smoke", Settings.EMPTY);
new DefaultIndexDefinition(adminClient, "test_smoke", "doc", Settings.EMPTY);
assertEquals("test_smoke", indexDefinition.getIndex());
assertTrue(indexDefinition.getFullIndexName().startsWith("test_smoke"));
assertEquals(0, indexDefinition.getReplicaLevel());
assertEquals(helper.getClusterName(), adminClient.getClusterName());
indexDefinition.setFullIndexName("test_smoke");
indexDefinition.setType("doc");
bulkClient.newIndex(indexDefinition);
bulkClient.index("test_smoke", "doc", "1", true, "{ \"name\" : \"Hello World\"}"); // single doc ingest
bulkClient.flush();
bulkClient.index(indexDefinition, "1", true, "{ \"name\" : \"Hello World\"}"); // single doc ingest
bulkClient.waitForResponses(30, TimeUnit.SECONDS);
adminClient.checkMapping("test_smoke");
bulkClient.update("test_smoke", "doc", "1", "{ \"name\" : \"Another name\"}");
bulkClient.delete("test_smoke", "doc", "1");
bulkClient.flush();
adminClient.checkMapping(indexDefinition);
bulkClient.update(indexDefinition, "1", "{ \"name\" : \"Another name\"}");
bulkClient.delete(indexDefinition, "1");
bulkClient.waitForResponses(30, TimeUnit.SECONDS);
bulkClient.index("test_smoke", "doc", "1", true, "{ \"name\" : \"Hello World\"}");
bulkClient.delete("test_smoke", "doc", "1");
bulkClient.flush();
bulkClient.index(indexDefinition, "1", true, "{ \"name\" : \"Hello World\"}");
bulkClient.delete(indexDefinition, "1");
bulkClient.waitForResponses(30, TimeUnit.SECONDS);
adminClient.deleteIndex("test_smoke");
adminClient.deleteIndex(indexDefinition);
bulkClient.newIndex(indexDefinition);
bulkClient.index(indexDefinition.getFullIndexName(), "doc", "1", true, "{ \"name\" : \"Hello World\"}");
bulkClient.flush();
bulkClient.index(indexDefinition, "1", true, "{ \"name\" : \"Hello World\"}");
bulkClient.waitForResponses(30, TimeUnit.SECONDS);
adminClient.updateReplicaLevel(indexDefinition, 2);
int replica = adminClient.getReplicaLevel(indexDefinition);

1
elx-transport/src/test/java/org/xbib/elx/transport/test/package-info.java
Unescape Escape View File

@ -1 +0,0 @@
package org.xbib.elx.transport.test;

2
elx-transport/src/test/resources/log4j2-test.xml
Unescape Escape View File

@ -6,7 +6,7 @@
</Console>
</appenders>
<Loggers>
<Root level="info">
<Root level="debug">
<AppenderRef ref="Console" />
</Root>
</Loggers>

2
gradle.properties
Unescape Escape View File

@ -1,6 +1,6 @@
group = org.xbib
name = elx
version = 2.2.1.31
version = 2.2.1.32
gradle.wrapper.version = 6.6.1
xbib-metrics.version = 2.1.0

Write Preview
Loading…
Cancel
Save