forked from opensearch-project/data-prepper
-
Notifications
You must be signed in to change notification settings - Fork 0
Commit
This commit does not belong to any branch on this repository, and may belong to a fork outside of the repository.
ENH: support custom index template for ES6 in opensearch sink (opense…
…arch-project#3061) Signed-off-by: George Chen <qchea@amazon.com>
- Loading branch information
1 parent
c753ba6
commit 8821005
Showing
21 changed files
with
1,136 additions
and
514 deletions.
There are no files selected for viewing
This file contains bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
This file contains bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
49 changes: 49 additions & 0 deletions
49
...ava/org/opensearch/dataprepper/plugins/sink/opensearch/index/ComposableIndexTemplate.java
This file contains bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
| Original file line number | Diff line number | Diff line change |
|---|---|---|
| @@ -0,0 +1,49 @@ | ||
| package org.opensearch.dataprepper.plugins.sink.opensearch.index; | ||
|
|
||
| import java.util.Collections; | ||
| import java.util.HashMap; | ||
| import java.util.List; | ||
| import java.util.Map; | ||
| import java.util.Optional; | ||
|
|
||
| public class ComposableIndexTemplate implements IndexTemplate { | ||
|
|
||
| private final Map<String, Object> indexTemplateMap; | ||
| private String name; | ||
|
|
||
| public ComposableIndexTemplate(final Map<String, Object> indexTemplateMap) { | ||
| this.indexTemplateMap = new HashMap<>(indexTemplateMap); | ||
| } | ||
|
|
||
| @Override | ||
| public void setTemplateName(final String name) { | ||
| this.name = name; | ||
|
|
||
| } | ||
|
|
||
| @Override | ||
| public void setIndexPatterns(final List<String> indexPatterns) { | ||
| indexTemplateMap.put("index_patterns", indexPatterns); | ||
| } | ||
|
|
||
| @Override | ||
| public void putCustomSetting(final String name, final Object value) { | ||
|
|
||
| } | ||
|
|
||
| @Override | ||
| public Optional<Long> getVersion() { | ||
| if(!indexTemplateMap.containsKey("version")) | ||
| return Optional.empty(); | ||
| final Number version = (Number) indexTemplateMap.get("version"); | ||
| return Optional.of(version.longValue()); | ||
| } | ||
|
|
||
| public Map<String, Object> getIndexTemplateMap() { | ||
| return Collections.unmodifiableMap(indexTemplateMap); | ||
| } | ||
|
|
||
| public String getName() { | ||
| return name; | ||
| } | ||
| } |
This file contains bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
89 changes: 89 additions & 0 deletions
89
...rg/opensearch/dataprepper/plugins/sink/opensearch/index/ComposableTemplateAPIWrapper.java
This file contains bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
| Original file line number | Diff line number | Diff line change |
|---|---|---|
| @@ -0,0 +1,89 @@ | ||
| package org.opensearch.dataprepper.plugins.sink.opensearch.index; | ||
|
|
||
| import com.fasterxml.jackson.databind.ObjectMapper; | ||
| import jakarta.json.stream.JsonParser; | ||
| import org.opensearch.client.json.JsonpDeserializer; | ||
| import org.opensearch.client.json.JsonpMapper; | ||
| import org.opensearch.client.json.ObjectBuilderDeserializer; | ||
| import org.opensearch.client.json.ObjectDeserializer; | ||
| import org.opensearch.client.opensearch.OpenSearchClient; | ||
| import org.opensearch.client.opensearch.indices.ExistsIndexTemplateRequest; | ||
| import org.opensearch.client.opensearch.indices.GetIndexTemplateRequest; | ||
| import org.opensearch.client.opensearch.indices.GetIndexTemplateResponse; | ||
| import org.opensearch.client.opensearch.indices.PutIndexTemplateRequest; | ||
| import org.opensearch.client.opensearch.indices.put_index_template.IndexTemplateMapping; | ||
| import org.opensearch.client.transport.endpoints.BooleanResponse; | ||
|
|
||
| import java.io.ByteArrayInputStream; | ||
| import java.io.IOException; | ||
| import java.nio.charset.StandardCharsets; | ||
| import java.util.Optional; | ||
|
|
||
| public class ComposableTemplateAPIWrapper implements IndexTemplateAPIWrapper<GetIndexTemplateResponse> { | ||
| private static final ObjectMapper OBJECT_MAPPER = new ObjectMapper(); | ||
| private final OpenSearchClient openSearchClient; | ||
|
|
||
| public ComposableTemplateAPIWrapper(final OpenSearchClient openSearchClient) { | ||
| this.openSearchClient = openSearchClient; | ||
| } | ||
|
|
||
| @Override | ||
| public void putTemplate(final IndexTemplate indexTemplate) throws IOException { | ||
| if(!(indexTemplate instanceof ComposableIndexTemplate)) { | ||
| throw new IllegalArgumentException("Unexpected indexTemplate provided to createTemplate."); | ||
| } | ||
|
|
||
| final ComposableIndexTemplate composableIndexTemplate = (ComposableIndexTemplate) indexTemplate; | ||
| final String indexTemplateString = OBJECT_MAPPER.writeValueAsString( | ||
| composableIndexTemplate.getIndexTemplateMap()); | ||
|
|
||
| final ByteArrayInputStream byteIn = new ByteArrayInputStream( | ||
| indexTemplateString.getBytes(StandardCharsets.UTF_8)); | ||
| final JsonpMapper mapper = openSearchClient._transport().jsonpMapper(); | ||
| final JsonParser parser = mapper.jsonProvider().createParser(byteIn); | ||
|
|
||
| final PutIndexTemplateRequest putIndexTemplateRequest = PutIndexTemplateRequestDeserializer | ||
| .getJsonpDeserializer(composableIndexTemplate.getName()) | ||
| .deserialize(parser, mapper); | ||
|
|
||
| openSearchClient.indices().putIndexTemplate(putIndexTemplateRequest); | ||
| } | ||
|
|
||
| @Override | ||
| public Optional<GetIndexTemplateResponse> getTemplate(final String indexTemplateName) throws IOException { | ||
| final ExistsIndexTemplateRequest existsRequest = new ExistsIndexTemplateRequest.Builder() | ||
| .name(indexTemplateName) | ||
| .build(); | ||
| final BooleanResponse existsResponse = openSearchClient.indices().existsIndexTemplate(existsRequest); | ||
|
|
||
| if (!existsResponse.value()) { | ||
| return Optional.empty(); | ||
| } | ||
|
|
||
| final GetIndexTemplateRequest getRequest = new GetIndexTemplateRequest.Builder() | ||
| .name(indexTemplateName) | ||
| .build(); | ||
| return Optional.of(openSearchClient.indices().getIndexTemplate(getRequest)); | ||
| } | ||
|
|
||
| private static class PutIndexTemplateRequestDeserializer { | ||
| private static void setupPutIndexTemplateRequestDeserializer(final ObjectDeserializer<PutIndexTemplateRequest.Builder> objectDeserializer) { | ||
|
|
||
| objectDeserializer.add(PutIndexTemplateRequest.Builder::name, JsonpDeserializer.stringDeserializer(), "name"); | ||
| objectDeserializer.add(PutIndexTemplateRequest.Builder::indexPatterns, JsonpDeserializer.arrayDeserializer(JsonpDeserializer.stringDeserializer()), | ||
| "index_patterns"); | ||
| objectDeserializer.add(PutIndexTemplateRequest.Builder::version, JsonpDeserializer.longDeserializer(), "version"); | ||
| objectDeserializer.add(PutIndexTemplateRequest.Builder::priority, JsonpDeserializer.integerDeserializer(), "priority"); | ||
| objectDeserializer.add(PutIndexTemplateRequest.Builder::composedOf, JsonpDeserializer.arrayDeserializer(JsonpDeserializer.stringDeserializer()), | ||
| "composed_of"); | ||
| objectDeserializer.add(PutIndexTemplateRequest.Builder::template, IndexTemplateMapping._DESERIALIZER, "template"); | ||
| } | ||
|
|
||
| static JsonpDeserializer<PutIndexTemplateRequest> getJsonpDeserializer(final String name) { | ||
| return ObjectBuilderDeserializer | ||
| .lazy( | ||
| () -> new PutIndexTemplateRequest.Builder().name(name), | ||
| PutIndexTemplateRequestDeserializer::setupPutIndexTemplateRequestDeserializer); | ||
| } | ||
| } | ||
| } |
Oops, something went wrong.