Skip to content

Add force merge step to Enrich Policy execution #41969

New issue

Have a question about this project? Sign up for a free GitHub account to open an issue and contact its maintainers and the community.

Sign up for GitHub

By clicking “Sign up for GitHub”, you agree to our terms of service and privacy statement. We’ll occasionally send you account related emails.

Already on GitHub? Sign in to your account

Merged
merged 4 commits into from
May 22, 2019
Merged

Add force merge step to Enrich Policy execution #41969

Show file tree
Hide file tree
Changes from all commits
Commits
Show all changes
4 commits
Select commit Hold shift + click to select a range
fdf1e50
Add step to forcemerge enrich index after reindex
jbaiera May 8, 2019
9e0c254
Merge branch 'enrich' into enrich-add-force-merge
jbaiera May 8, 2019
61d6b9e
Merge branch 'enrich' into enrich-add-force-merge
jbaiera May 20, 2019
74fdd40
Merge branch 'enrich' into enrich-add-force-merge
jbaiera May 22, 2019
File filter

Filter by extension

Filter by extension
Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
23 changes: 20 additions & 3 deletions x-pack/plugin/enrich/src/main/java/org/elasticsearch/xpack/enrich/EnrichPolicyRunner.java
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -21,6 +21,8 @@
import org.elasticsearch.action.admin.indices.alias.get.GetAliasesRequest;
import org.elasticsearch.action.admin.indices.create.CreateIndexRequest;
import org.elasticsearch.action.admin.indices.create.CreateIndexResponse;
import org.elasticsearch.action.admin.indices.forcemerge.ForceMergeRequest;
import org.elasticsearch.action.admin.indices.forcemerge.ForceMergeResponse;
import org.elasticsearch.action.admin.indices.get.GetIndexRequest;
import org.elasticsearch.action.admin.indices.get.GetIndexResponse;
import org.elasticsearch.action.admin.indices.refresh.RefreshRequest;
Expand Down Expand Up @@ -107,14 +109,14 @@ private void validateMappings(final GetIndexResponse getIndexResponse) {
logger.debug("Policy [{}]: Validating [{}] source mappings", policyName, sourceIndices);
for (String sourceIndex : sourceIndices) {
Map<String, Object> mapping = getMappings(getIndexResponse, sourceIndex);
Set<?> properties = ((Map<?, ?>) mapping.get("properties")).keySet();
Map<?, ?> properties = ((Map<?, ?>) mapping.get("properties"));
if (properties == null) {
listener.onFailure(
new ElasticsearchException(
"Enrich policy execution for [{}] failed. Could not read mapping for source [{}] included by pattern [{}]",
policyName, sourceIndex, policy.getIndices()));
}
if (properties.contains(policy.getEnrichKey()) == false) {
if (properties.containsKey(policy.getEnrichKey()) == false) {
listener.onFailure(
new ElasticsearchException(
"Enrich policy execution for [{}] failed. Could not locate enrich key field [{}] on mapping for index [{}]",
Expand Down Expand Up @@ -209,7 +211,7 @@ public void onResponse(BulkByScrollResponse bulkByScrollResponse) {
} else {
logger.info("Policy [{}]: Transferred [{}] documents to enrich index [{}]", policyName,
bulkByScrollResponse.getCreated(), destinationIndexName);
refreshEnrichIndex(destinationIndexName);
forceMergeEnrichIndex(destinationIndexName);
}
}

Expand All @@ -220,6 +222,21 @@ public void onFailure(Exception e) {
});
}

private void forceMergeEnrichIndex(final String destinationIndexName) {
logger.debug("Policy [{}]: Force merging newly created enrich index [{}]", policyName, destinationIndexName);
client.admin().indices().forceMerge(new ForceMergeRequest(destinationIndexName).maxNumSegments(1), new ActionListener<>() {
@Override
public void onResponse(ForceMergeResponse forceMergeResponse) {
refreshEnrichIndex(destinationIndexName);
}

@Override
public void onFailure(Exception e) {
listener.onFailure(e);
}
});
}

private void refreshEnrichIndex(final String destinationIndexName) {
logger.debug("Policy [{}]: Refreshing newly created enrich index [{}]", policyName, destinationIndexName);
client.admin().indices().refresh(new RefreshRequest(destinationIndexName), new ActionListener<>() {
Expand Down
42 changes: 42 additions & 0 deletions ...k/plugin/enrich/src/test/java/org/elasticsearch/xpack/enrich/EnrichPolicyRunnerTests.java
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -16,6 +16,11 @@
import org.elasticsearch.action.ActionListener;
import org.elasticsearch.action.admin.indices.get.GetIndexRequest;
import org.elasticsearch.action.admin.indices.get.GetIndexResponse;
import org.elasticsearch.action.admin.indices.segments.IndexSegments;
import org.elasticsearch.action.admin.indices.segments.IndexShardSegments;
import org.elasticsearch.action.admin.indices.segments.IndicesSegmentResponse;
import org.elasticsearch.action.admin.indices.segments.IndicesSegmentsRequest;
import org.elasticsearch.action.admin.indices.segments.ShardSegments;
import org.elasticsearch.action.index.IndexRequest;
import org.elasticsearch.action.index.IndexResponse;
import org.elasticsearch.action.search.SearchRequest;
Expand All @@ -25,6 +30,7 @@
import org.elasticsearch.cluster.service.ClusterService;
import org.elasticsearch.common.settings.Settings;
import org.elasticsearch.common.xcontent.XContentType;
import org.elasticsearch.index.engine.Segment;
import org.elasticsearch.index.query.QueryBuilders;
import org.elasticsearch.index.reindex.ReindexPlugin;
import org.elasticsearch.plugins.Plugin;
Expand Down Expand Up @@ -117,6 +123,7 @@ public void onFailure(Exception e) {
throw exception.get();
}

// Validate Index definition
String createdEnrichIndex = ".enrich-test1-" + createTime;
GetIndexResponse enrichIndex = client().admin().indices()
.getIndex(new GetIndexRequest().indices(".enrich-test1")).get();
Expand All @@ -125,6 +132,8 @@ public void onFailure(Exception e) {
Settings settings = enrichIndex.getSettings().get(createdEnrichIndex);
assertNotNull(settings);
assertThat(settings.get("index.auto_expand_replicas"), is(equalTo("0-all")));

// Validate Mapping
Map<String, Object> mapping = enrichIndex.getMappings().get(createdEnrichIndex).get("_doc").sourceAsMap();
assertThat(mapping.get("dynamic"), is("false"));
Map<?, ?> properties = (Map<?, ?>) mapping.get("properties");
Expand All @@ -135,6 +144,7 @@ public void onFailure(Exception e) {
assertThat(field1.get("type"), is(equalTo("keyword")));
assertThat(field1.get("doc_values"), is(false));

// Validate document structure
SearchResponse enrichSearchResponse = client().search(
new SearchRequest(".enrich-test1")
.source(SearchSourceBuilder.searchSource()
Expand All @@ -147,6 +157,20 @@ public void onFailure(Exception e) {
assertThat(enrichDocument.get("field1"), is(equalTo("value1")));
assertThat(enrichDocument.get("field2"), is(equalTo(2)));
assertThat(enrichDocument.get("field5"), is(equalTo("value5")));

// Validate segments
IndicesSegmentResponse indicesSegmentResponse = client().admin().indices()
.segments(new IndicesSegmentsRequest(createdEnrichIndex)).get();
IndexSegments indexSegments = indicesSegmentResponse.getIndices().get(createdEnrichIndex);
assertNotNull(indexSegments);
assertThat(indexSegments.getShards().size(), is(equalTo(1)));
IndexShardSegments shardSegments = indexSegments.getShards().get(0);
assertNotNull(shardSegments);
assertThat(shardSegments.getShards().length, is(equalTo(1)));
ShardSegments shard = shardSegments.getShards()[0];
assertThat(shard.getSegments().size(), is(equalTo(1)));
Segment segment = shard.getSegments().iterator().next();
assertThat(segment.getNumDocs(), is(equalTo(1)));
}

public void testRunnerMultiSource() throws Exception {
Expand Down Expand Up @@ -233,13 +257,16 @@ public void onFailure(Exception e) {
throw exception.get();
}

// Validate Index definition
String createdEnrichIndex = ".enrich-test1-" + createTime;
GetIndexResponse enrichIndex = client().admin().indices().getIndex(new GetIndexRequest().indices(".enrich-test1")).actionGet();
assertThat(enrichIndex.getIndices().length, equalTo(1));
assertThat(enrichIndex.getIndices()[0], equalTo(createdEnrichIndex));
Settings settings = enrichIndex.getSettings().get(createdEnrichIndex);
assertNotNull(settings);
assertThat(settings.get("index.auto_expand_replicas"), is(equalTo("0-all")));

// Validate Mapping
Map<String, Object> mapping = enrichIndex.getMappings().get(createdEnrichIndex).get("_doc").sourceAsMap();
assertThat(mapping.get("dynamic"), is("false"));
Map<?, ?> properties = (Map<?, ?>) mapping.get("properties");
Expand All @@ -250,6 +277,7 @@ public void onFailure(Exception e) {
assertThat(field1.get("type"), is(equalTo("keyword")));
assertThat(field1.get("doc_values"), is(false));

// Validate document structure
SearchResponse enrichSearchResponse = client().search(
new SearchRequest(".enrich-test1")
.source(SearchSourceBuilder.searchSource()
Expand All @@ -262,5 +290,19 @@ public void onFailure(Exception e) {
assertThat(enrichDocument.get("field1"), is(equalTo("value1")));
assertThat(enrichDocument.get("field2"), is(equalTo(2)));
assertThat(enrichDocument.get("field5"), is(equalTo("value5")));

// Validate segments
IndicesSegmentResponse indicesSegmentResponse = client().admin().indices()
.segments(new IndicesSegmentsRequest(createdEnrichIndex)).get();
IndexSegments indexSegments = indicesSegmentResponse.getIndices().get(createdEnrichIndex);
assertNotNull(indexSegments);
assertThat(indexSegments.getShards().size(), is(equalTo(1)));
IndexShardSegments shardSegments = indexSegments.getShards().get(0);
assertNotNull(shardSegments);
assertThat(shardSegments.getShards().length, is(equalTo(1)));
ShardSegments shard = shardSegments.getShards()[0];
assertThat(shard.getSegments().size(), is(equalTo(1)));
Segment segment = shard.getSegments().iterator().next();
assertThat(segment.getNumDocs(), is(equalTo(3)));
}
}