From 4a1d7b2ff4ff3d4e75035de04943eedf977278b3 Mon Sep 17 00:00:00 2001 From: p-kimberley Date: Wed, 2 Aug 2023 19:42:14 +1000 Subject: [PATCH 1/2] Back-port #3453 to 7.1 --- .../indexing/ElasticIndexingFilter.java | 268 ++++++++---------- .../20230515_151828_902__3453.md | 24 ++ 2 files changed, 140 insertions(+), 152 deletions(-) create mode 100644 unreleased_changes/20230515_151828_902__3453.md diff --git a/stroom-search/stroom-search-elastic/src/main/java/stroom/search/elastic/indexing/ElasticIndexingFilter.java b/stroom-search/stroom-search-elastic/src/main/java/stroom/search/elastic/indexing/ElasticIndexingFilter.java index a176207debf..ab23ef3e191 100644 --- a/stroom-search/stroom-search-elastic/src/main/java/stroom/search/elastic/indexing/ElasticIndexingFilter.java +++ b/stroom-search/stroom-search-elastic/src/main/java/stroom/search/elastic/indexing/ElasticIndexingFilter.java @@ -42,7 +42,6 @@ import stroom.util.logging.LambdaLogger; import stroom.util.logging.LambdaLoggerFactory; import stroom.util.shared.Severity; -import stroom.util.time.StroomDuration; import com.fasterxml.jackson.core.JsonFactory; import com.fasterxml.jackson.core.JsonGenerator; @@ -76,18 +75,16 @@ import java.io.ByteArrayOutputStream; import java.io.IOException; import java.security.InvalidParameterException; -import java.text.SimpleDateFormat; -import java.time.ZoneId; -import java.time.ZonedDateTime; -import java.time.format.DateTimeFormatter; -import java.time.format.DateTimeParseException; -import java.time.temporal.ChronoUnit; import java.util.ArrayList; -import java.util.Date; +import java.util.HashMap; +import java.util.HashSet; import java.util.List; import java.util.Map; -import java.util.TimeZone; +import java.util.Set; import java.util.concurrent.atomic.AtomicBoolean; +import java.util.regex.Matcher; +import java.util.regex.Pattern; +import java.util.stream.Collectors; import javax.inject.Inject; import javax.inject.Provider; import javax.ws.rs.NotFoundException; @@ -106,6 +103,8 @@ class ElasticIndexingFilter extends AbstractXMLFilter { private static final int INITIAL_JSON_STREAM_SIZE_BYTES = 1024; private static final int ES_MAX_EXCEPTION_CHARS = 1024; private static final String ES_TOO_MANY_REQUESTS_STATUS = "TOO_MANY_REQUESTS"; + private static final Pattern INDEX_NAME_VALUE_PATTERN = Pattern.compile("(\\{[^}]+?})"); + private static final Pattern INDEX_BASE_NAME_PATTERN = Pattern.compile("^([^{]+)"); // Dependencies private final LocationFactoryProxy locationFactory; @@ -123,12 +122,7 @@ class ElasticIndexingFilter extends AbstractXMLFilter { private String ingestPipelineName = null; private boolean refreshAfterEachBatch = false; private DocRef clusterRef; - private String indexBaseName; - private String indexNameDateFormat; - private String indexNameDateFieldName = "@timestamp"; - private Date indexNameDateMin; - private SimpleDateFormat indexNameDateMinFormat; - private StroomDuration indexNameDateMaxFutureOffset; + private String indexName; // Cached entities ElasticClusterDoc elasticCluster; @@ -138,11 +132,12 @@ class ElasticIndexingFilter extends AbstractXMLFilter { private final List indexRequests; private final ByteArrayOutputStream currentDocument; private final StringBuilder valueBuffer = new StringBuilder(); + private Set indexNameVariables = new HashSet<>(); + private final Map currentDocIndexNameVariables = new HashMap<>(); private String currentDocFieldName = null; private int currentDocPropertyCount = 0; private boolean inOuterArray = false; private int currentDepth = 0; - private String currentDocTimestamp = null; private JsonGenerator jsonGenerator; private int currentRetry; @@ -170,10 +165,6 @@ class ElasticIndexingFilter extends AbstractXMLFilter { indexRequests = new ArrayList<>(); currentDocument = new ByteArrayOutputStream(INITIAL_JSON_STREAM_SIZE_BYTES); currentRetry = 0; - - indexNameDateMinFormat = new SimpleDateFormat("yyyy"); - indexNameDateMinFormat.setTimeZone(TimeZone.getTimeZone("UTC)")); - indexNameDateMaxFutureOffset = StroomDuration.parse("P1D"); } /** @@ -186,7 +177,7 @@ public void startProcessing() { fatalError("Elasticsearch cluster ref has not been set", new NotFoundException()); } - if (indexBaseName == null || indexBaseName.isEmpty()) { + if (indexName == null || indexName.isEmpty()) { fatalError("Index name has not been set", new InvalidParameterException()); } @@ -215,6 +206,8 @@ public void startProcessing() { jsonGenerator = JSON_FACTORY.createGenerator(currentDocument); + populateIndexNameVariableNames(); + } catch (IOException e) { fatalError("Failed to initialise JsonGenerator", e); } finally { @@ -222,6 +215,15 @@ public void startProcessing() { } } + private void populateIndexNameVariableNames() { + indexNameVariables = INDEX_NAME_VALUE_PATTERN.matcher(indexName).results() + .map(matchResult -> { + final String fieldNameMatch = matchResult.group(); + return fieldNameMatch.substring(1, fieldNameMatch.length() - 1); + }) + .collect(Collectors.toSet()); + } + @Override public void endProcessing() { try { @@ -268,6 +270,9 @@ public void startElement(final String uri, final String localName, final String incrementDepth(); writeFieldName(); jsonGenerator.writeStartObject(); + if (currentDepth == 1) { + enterDocumentRoot(); + } } catch (IOException e) { fatalError("Invalid start of object", e); } @@ -290,6 +295,10 @@ public void startElement(final String uri, final String localName, final String super.startElement(uri, localName, qName, attributes); } + private void enterDocumentRoot() { + currentDocIndexNameVariables.clear(); + } + private void incrementDepth() { final int maxDepth = elasticConfigProvider.get().getIndexingConfig().getMaxNestedElementDepth(); @@ -338,14 +347,13 @@ public void endElement(final String uri, final String localName, final String qN case JSONParser.XML_ELEMENT_STRING: value = valueBuffer.toString(); try { - if (value.length() > 0) { - writeFieldName(); - jsonGenerator.writeString(value); - currentDocPropertyCount++; - } - if (currentDocFieldName != null && currentDocFieldName.equals(indexNameDateFieldName)) { - // This is the timestamp field, so store its value for formatting the full index name - currentDocTimestamp = value; + if (!value.isEmpty()) { + if (includeField(currentDocFieldName)) { + writeFieldName(); + jsonGenerator.writeString(value); + currentDocPropertyCount++; + } + storeIndexNameVariableValue(value); } } catch (IOException e) { fatalError("Invalid string value '" + value + "' for property '" + @@ -355,10 +363,13 @@ public void endElement(final String uri, final String localName, final String qN case JSONParser.XML_ELEMENT_BOOLEAN: value = valueBuffer.toString(); try { - if (value.length() > 0) { - writeFieldName(); - jsonGenerator.writeBoolean(Boolean.parseBoolean(value)); - currentDocPropertyCount++; + if (!value.isEmpty()) { + if (includeField(currentDocFieldName)) { + writeFieldName(); + jsonGenerator.writeBoolean(Boolean.parseBoolean(value)); + currentDocPropertyCount++; + } + storeIndexNameVariableValue(value); } } catch (IOException e) { fatalError("Invalid boolean value '" + value + "' for property '" + @@ -367,9 +378,12 @@ public void endElement(final String uri, final String localName, final String qN break; case JSONParser.XML_ELEMENT_NULL: try { - writeFieldName(); - jsonGenerator.writeNull(); - currentDocPropertyCount++; + if (includeField(currentDocFieldName)) { + writeFieldName(); + jsonGenerator.writeNull(); + currentDocPropertyCount++; + } + storeIndexNameVariableValue(""); } catch (IOException e) { fatalError("Invalid null value for property '" + currentDocFieldName + "'", e); } @@ -377,10 +391,13 @@ public void endElement(final String uri, final String localName, final String qN case JSONParser.XML_ELEMENT_NUMBER: value = valueBuffer.toString(); try { - if (value.length() > 0) { - writeFieldName(); - jsonGenerator.writeNumber(value); - currentDocPropertyCount++; + if (!value.isEmpty()) { + if (includeField(currentDocFieldName)) { + writeFieldName(); + jsonGenerator.writeNumber(value); + currentDocPropertyCount++; + } + storeIndexNameVariableValue(value); } } catch (IOException e) { fatalError("Invalid number value '" + value + "' for property '" + @@ -396,6 +413,25 @@ public void endElement(final String uri, final String localName, final String qN super.endElement(uri, localName, qName); } + /** + * Whether to include the field in the destination document + */ + private boolean includeField(final String fieldName) { + return fieldName == null || !fieldName.startsWith("_"); + } + + private void storeIndexNameVariableValue(final String value) { + try { + if (!value.isEmpty() && indexNameVariables.contains(currentDocFieldName)) { + currentDocIndexNameVariables.put(currentDocFieldName, value); + } + } catch (IllegalArgumentException e) { + fatalError("Index variable '" + currentDocFieldName + "' specified more than once in document", e); + } catch (Exception e) { + fatalError("Unexpected error parsing index name variable value", e); + } + } + /** * @param ch the characters from the XML document * @param start the start position in the array @@ -456,7 +492,6 @@ private void processDocument() { private void clearDocument() { // Discard the document currentDocument.reset(); - currentDocTimestamp = null; // Reset the count of how many fields we have indexed for the current event currentDocPropertyCount = 0; @@ -522,6 +557,7 @@ private List getTargetIndexNames(final RestHighLevelClient elasticClient .size(100) // Number of index names to retrieve at a time, until we have all of them .query(new TermQueryBuilder(ElasticIndexConstants.STREAM_ID, streamId)) .aggregation(compositeAgg); + final String indexBaseName = getIndexBaseName(); final SearchRequest searchRequest = new SearchRequest(indexBaseName + "*") .source(searchSourceBuilder); final SearchResponse searchResponse = elasticClient.search(searchRequest, RequestOptions.DEFAULT); @@ -542,11 +578,24 @@ private List getTargetIndexNames(final RestHighLevelClient elasticClient return indexNames; } catch (IOException e) { fatalError("Failed to list indices for reindex purge. StreamId: " + streamId + ". " + - "Base name: '" + indexBaseName + "'", e); + "Base name: '" + indexName + "'", e); return null; } } + /** + * Get the index name stem, which is the text up to the first index name variable, as denoted by curly braces + */ + private String getIndexBaseName() { + final Matcher indexNameMatcher = INDEX_BASE_NAME_PATTERN.matcher(indexName); + if (indexNameMatcher.find()) { + return indexNameMatcher.group(1); + } else { + throw new RuntimeException("Expected one or more characters in the index name before the first curly " + + "brace."); + } + } + /** * Index the current batch of documents */ @@ -615,7 +664,6 @@ private void indexDocuments() { * Elasticsearch rejected the indexing request, because it is overloaded and * cannot queue the batched payload. Retry after a delay, if we haven't exceeded the retry * limit. Otherwise, terminate this thread and create an `Error` stream. - * @param e */ private void handleElasticsearchException(final RuntimeException e) { String statusName = ""; @@ -650,49 +698,28 @@ private void handleElasticsearchException(final RuntimeException e) { } /** - * Where an index name date pattern is specified, formats the value of the document timestamp field - * using the date pattern. Otherwise, the index base name is returned. - * @return Index base name appended with the formatted document timestamp + * Build the index name, substituting values from the source document where specified. + * For instance, consider an index name of `stroom-index-{_year}`. A field `2023` in + * the doc root will result in a destination index of `stroom-index-2023` for that document. In this case, because + * the field `_year` starts with an underscore, it is not rendered to the doc and instead is solely used in the + * index name. */ private String formatIndexName() { - // If a date format has specified, append the formatted timestamp field - // value to the index base name - if (currentDocTimestamp == null || indexNameDateFormat == null || indexNameDateFormat.isEmpty()) { - return indexBaseName; - } else { - try { - final ZonedDateTime eventTime = ZonedDateTime.parse(currentDocTimestamp); - final DateTimeFormatter pattern = DateTimeFormatter.ofPattern(indexNameDateFormat); - - // If the event occurred after now + max offset, don't append a time suffix to the index name - if (!indexNameDateMaxFutureOffset.isZero()) { - final ZonedDateTime timeNow = ZonedDateTime.now(eventTime.getZone()); - final ZonedDateTime nowPlusOffset = timeNow.plusSeconds( - indexNameDateMaxFutureOffset.get(ChronoUnit.SECONDS)); - if (eventTime.isAfter(nowPlusOffset)) { - return indexBaseName; - } - } - - // If the event occurred before the minimum date, don't append a time suffix to the index name - if (indexNameDateMin != null) { - if (eventTime.isBefore(indexNameDateMin.toInstant().atZone(ZoneId.of("UTC")))) { - return indexBaseName; - } - } - - return indexBaseName + pattern.format(eventTime); - } catch (IllegalArgumentException e) { - throw new RuntimeException(String.format("Invalid index name date format: %s", indexNameDateFormat)); - } catch (Exception e) { - throw new NotFoundException(String.format("Field %s not found in document, which is " + - "required as property `indexNameDateFormat` has been set", indexNameDateFieldName)); + final Matcher indexNameVariableMatcher = INDEX_NAME_VALUE_PATTERN.matcher(indexName); + return indexNameVariableMatcher.replaceAll(matchResult -> { + final String fieldNameMatch = matchResult.group(); + final String fieldName = fieldNameMatch.substring(1, fieldNameMatch.length() - 1); + final String fieldValue = currentDocIndexNameVariables.get(fieldName); + if (fieldValue == null) { + throw new IllegalArgumentException("Field '" + fieldName + "' not found in document when " + + "building index name with pattern: '" + indexName + "'"); } - } + return fieldValue; + }); } @PipelineProperty( - description = "Target Elasticsearch cluster", + description = "Target Elasticsearch cluster.", displayPriority = 1 ) @PipelinePropertyDocRef(types = ElasticClusterDoc.DOCUMENT_TYPE) @@ -701,7 +728,7 @@ public void setCluster(final DocRef clusterRef) { } @PipelineProperty( - description = "Maximum number of documents to index in each bulk request", + description = "Maximum number of documents to index in each bulk request.", defaultValue = "10000", displayPriority = 2 ) @@ -711,7 +738,7 @@ public void setBatchSize(final int batchSize) { @PipelineProperty( description = "Refresh the index after each batch is processed, making the indexed documents visible to " + - "searches", + "searches.", defaultValue = "false", displayPriority = 3 ) @@ -720,7 +747,7 @@ public void setRefreshAfterEachBatch(final boolean refreshAfterEachBatch) { } @PipelineProperty( - description = "Name of the Elasticsearch ingest pipeline to execute when indexing", + description = "Name of the Elasticsearch ingest pipeline to execute when indexing.", displayPriority = 4 ) public void setIngestPipeline(final String ingestPipelineName) { @@ -728,81 +755,18 @@ public void setIngestPipeline(final String ingestPipelineName) { } @PipelineProperty( - description = "Name of the Elasticsearch index", + description = "Name of the Elasticsearch index. Variables specified such as `{year}` are replaced with " + + "the corresponding field values contained in the document root. Field names beginning with an " + + "underscore are not written to the document and are only used in the index name pattern.", displayPriority = 5 ) - public void setIndexBaseName(final String indexBaseName) { - this.indexBaseName = indexBaseName; - } - - @PipelineProperty( - description = "Format of the date to append to the index name (example: `-yyyy`). If unspecified, no " + - "date is appended.", - displayPriority = 6 - ) - public void setIndexNameDateFormat(final String indexNameDateFormat) { - this.indexNameDateFormat = indexNameDateFormat; - } - - @PipelineProperty( - description = "Name of the field containing the `DateTime` value to use when determining the index date " + - "suffix", - defaultValue = "@timestamp", - displayPriority = 7 - ) - public void setIndexNameDateFieldName(final String indexNameDateFieldName) { - this.indexNameDateFieldName = indexNameDateFieldName; - } - - @PipelineProperty( - description = "Do not append a time suffix to the index name for events occurring before this date. " + - "Date is assumed to be in UTC and of the format specified in `indexNameDateMinFormat`", - displayPriority = 8 - ) - public void setIndexNameDateMin(final String indexNameDateMin) { - try { - if (indexNameDateMin != null && !indexNameDateMin.isEmpty() && indexNameDateMinFormat != null) { - this.indexNameDateMin = indexNameDateMinFormat.parse(indexNameDateMin); - } else { - this.indexNameDateMin = null; - } - } catch (DateTimeParseException e) { - this.indexNameDateMin = null; - fatalError("Invalid value " + indexNameDateMin + " provided for indexNameDateMin", e); - } catch (Exception e) { - fatalError("Error occurred when parsing date value: " + indexNameDateMin, e); - } - } - - @PipelineProperty( - description = "Date format of the supplied `indexNameDateMin` property", - defaultValue = "yyyy", - displayPriority = 9 - ) - public void setIndexNameDateMinFormat(final String indexNameDateMinFormat) { - try { - this.indexNameDateMinFormat = new SimpleDateFormat(indexNameDateMinFormat); - this.indexNameDateMinFormat.setTimeZone(TimeZone.getTimeZone("UTC")); - } catch (IllegalArgumentException e) { - this.indexNameDateMinFormat = null; - fatalError("Invalid date format " + indexNameDateMinFormat + " provided for " + - "indexNameDateMinFormat", e); - } - } - - @PipelineProperty( - description = "Do not append a time suffix to the index name for events occurring after the current " + - "time plus the specified offset", - defaultValue = "P1D", - displayPriority = 10 - ) - public void setIndexNameDateMaxFutureOffset(final String indexNameDateMaxFutureOffset) { - this.indexNameDateMaxFutureOffset = StroomDuration.parse(indexNameDateMaxFutureOffset); + public void setIndexName(final String indexName) { + this.indexName = indexName; } @PipelineProperty( description = "When reprocessing a stream, first delete any documents from the index matching the " + - "stream ID", + "source stream ID.", defaultValue = "true", displayPriority = 11 ) diff --git a/unreleased_changes/20230515_151828_902__3453.md b/unreleased_changes/20230515_151828_902__3453.md new file mode 100644 index 00000000000..82c3ec7f5dd --- /dev/null +++ b/unreleased_changes/20230515_151828_902__3453.md @@ -0,0 +1,24 @@ +* Issue **#3453** : Support Elasticsearch index name variable substitution. + + +```sh +# ******************************************************************************** +# Issue title: Allow override of destination Elasticsearch index on a per-doc basis +# Issue link: https://github.com/gchq/stroom/issues/3453 +# ******************************************************************************** + +# ONLY the top line will be included as a change entry in the CHANGELOG. +# The entry should be in GitHub flavour markdown and should be written on a SINGLE +# line with no hard breaks. You can have multiple change files for a single GitHub issue. +# The entry should be written in the imperative mood, i.e. 'Fix nasty bug' rather than +# 'Fixed nasty bug'. +# +# Examples of acceptable entries are: +# +# +# * Issue **123** : Fix bug with an associated GitHub issue in this repository +# +# * Issue **namespace/other-repo#456** : Fix bug with an associated GitHub issue in another repository +# +# * Fix bug with no associated GitHub issue. +``` From 570a07091e9cba936a2723b40244dda4b898679e Mon Sep 17 00:00:00 2001 From: p-kimberley Date: Wed, 2 Aug 2023 22:55:32 +1000 Subject: [PATCH 2/2] Bump ES client version --- build.gradle | 2 +- 1 file changed, 1 insertion(+), 1 deletion(-) diff --git a/build.gradle b/build.gradle index bd3de896d41..c973b2bc2bb 100644 --- a/build.gradle +++ b/build.gradle @@ -112,7 +112,7 @@ ext.versions = [ aws : '2.17.253', curator : '4.2.0', // Curator 4 works with ZK 3.4.x in soft compatibility mode, i.e. you must exlude its dep on ZK and explicitly add one for 3.4.x dropwizard : '2.0.28', // used to set the dropwizard-bom version, that controls lots of dependency versions - elasticsearch : '7.17.8', + elasticsearch : '7.17.12', httpcore : '4.4.12', // Transient dependency of Elasticsearch guice5 : '5.1.0', gwt : '2.9.0',