ingest: Introduction of a bytes processor

docs/reference/ingest/ingest-node.asciidoc

@@ -773,6 +773,33 @@ Accepts a single value or an array of values.
 --------------------------------------------------
 // NOTCONSOLE
 
+[[bytes-processor]]
+=== Bytes Processor
+Converts a human readable byte value (e.g. 1kb) to its value in bytes (e.g. 1024).
+
+Supported human readable units are "b", "kb", "mb", "gb", "tb", "pb" case insensitive. An error will occur if
+the field is not a supported format or resultant value exceeds 2^63.
+
+[[bytes-options]]
+.Bytes Options
+[options="header"]
+|======
+| Name | Required | Default | Description
+| `field` | yes | - | The field to convert
+| `target_field` | no | `field` | The field to assign the converted value to, by default `field` is updated in-place
+| `ignore_missing` | no | `false` | If `true` and `field` does not exist or is `null`, the processor quietly exits without modifying the document
+|======
+
+[source,js]
+--------------------------------------------------
+{
+ "bytes": {
+ "field": "foo"
+ }
+}
+--------------------------------------------------
+// NOTCONSOLE
+
 [[convert-processor]]
 === Convert Processor
 Converts an existing field's value to a different type, such as converting a string to an integer.

modules/ingest-common/src/main/java/org/elasticsearch/ingest/common/AbstractStringProcessor.java

@@ -27,10 +27,12 @@
 import java.util.Map;
 
 /**
- * Base class for processors that manipulate strings and require a single "fields" array config value, which
+ * Base class for processors that manipulate source strings and require a single "fields" array config value, which
  * holds a list of field names in string format.
+ *
+ * @param <T> The resultant type for the target field
  */
-abstract class AbstractStringProcessor extends AbstractProcessor {
+abstract class AbstractStringProcessor<T> extends AbstractProcessor {
  private final String field;
  private final boolean ignoreMissing;
  private final String targetField;
@@ -67,7 +69,7 @@ public final void execute(IngestDocument document) {
  document.setFieldValue(targetField, process(val));
  }
 
- protected abstract String process(String value);
+ protected abstract T process(String value);
 
  abstract static class Factory implements Processor.Factory {
  final String processorType;

modules/ingest-common/src/main/java/org/elasticsearch/ingest/common/BytesProcessor.java

@@ -0,0 +1,60 @@
+/*
+ * Licensed to Elasticsearch under one or more contributor
+ * license agreements. See the NOTICE file distributed with
+ * this work for additional information regarding copyright
+ * ownership. Elasticsearch licenses this file to you under
+ * the Apache License, Version 2.0 (the "License"); you may
+ * not use this file except in compliance with the License.
+ * You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing,
+ * software distributed under the License is distributed on an
+ * "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
+ * KIND, either express or implied. See the License for the
+ * specific language governing permissions and limitations
+ * under the License.
+ */
+
+package org.elasticsearch.ingest.common;
+
+import org.elasticsearch.common.unit.ByteSizeValue;
+
+import java.util.Map;
+
+/**
+ * Processor that converts the content of string fields to the byte value.
+ * Throws exception is the field is not of type string or can not convert to the numeric byte value
+ */
+public final class BytesProcessor extends AbstractStringProcessor {
+
+ public static final String TYPE = "bytes";
+
+ BytesProcessor(String processorTag, String field, boolean ignoreMissing, String targetField) {
+ super(processorTag, field, ignoreMissing, targetField);
+ }
+
+ @Override
+ protected Long process(String value) {
+ return ByteSizeValue.parseBytesSizeValue(value, null, getField()).getBytes();
+ }
+
+ @Override
+ public String getType() {
+ return TYPE;
+ }
+
+ public static final class Factory extends AbstractStringProcessor.Factory {
+
+ public Factory() {
+ super(TYPE);
+ }
+
+ @Override
+ protected BytesProcessor newProcessor(String tag, Map<String, Object> config, String field,
+ boolean ignoreMissing, String targetField) {
+ return new BytesProcessor(tag, field, ignoreMissing, targetField);
+ }
+ }
+}

modules/ingest-common/src/main/java/org/elasticsearch/ingest/common/IngestCommonPlugin.java

@@ -81,6 +81,7 @@ public Map<String, Processor.Factory> getProcessors(Processor.Parameters paramet
  processors.put(JsonProcessor.TYPE, new JsonProcessor.Factory());
  processors.put(KeyValueProcessor.TYPE, new KeyValueProcessor.Factory());
  processors.put(URLDecodeProcessor.TYPE, new URLDecodeProcessor.Factory());
+ processors.put(BytesProcessor.TYPE, new BytesProcessor.Factory());
  return Collections.unmodifiableMap(processors);
  }
 

modules/ingest-common/src/test/java/org/elasticsearch/ingest/common/AbstractStringProcessorTestCase.java

@@ -31,23 +31,27 @@
 import static org.hamcrest.Matchers.containsString;
 import static org.hamcrest.Matchers.equalTo;
 
-public abstract class AbstractStringProcessorTestCase extends ESTestCase {
+public abstract class AbstractStringProcessorTestCase<T> extends ESTestCase {
 
  protected abstract AbstractStringProcessor newProcessor(String field, boolean ignoreMissing, String targetField);
 
  protected String modifyInput(String input) {
  return input;
  }
 
- protected abstract String expectedResult(String input);
+ protected abstract T expectedResult(String input);
+
+ protected Class<T> expectedResultType(){
+ return (Class<T>) String.class; // most results types are Strings
+ }
 
  public void testProcessor() throws Exception {
  IngestDocument ingestDocument = RandomDocumentPicks.randomIngestDocument(random());
  String fieldValue = RandomDocumentPicks.randomString(random());
  String fieldName = RandomDocumentPicks.addRandomField(random(), ingestDocument, modifyInput(fieldValue));
  Processor processor = newProcessor(fieldName, randomBoolean(), fieldName);
  processor.execute(ingestDocument);
- assertThat(ingestDocument.getFieldValue(fieldName, String.class), equalTo(expectedResult(fieldValue)));
+ assertThat(ingestDocument.getFieldValue(fieldName, expectedResultType()), equalTo(expectedResult(fieldValue)));
  }
 
  public void testFieldNotFound() throws Exception {
@@ -109,6 +113,6 @@ public void testTargetField() throws Exception {
  String targetFieldName = fieldName + "foo";
  Processor processor = newProcessor(fieldName, randomBoolean(), targetFieldName);
  processor.execute(ingestDocument);
- assertThat(ingestDocument.getFieldValue(targetFieldName, String.class), equalTo(expectedResult(fieldValue)));
+ assertThat(ingestDocument.getFieldValue(targetFieldName, expectedResultType()), equalTo(expectedResult(fieldValue)));
  }
 }

modules/ingest-common/src/test/java/org/elasticsearch/ingest/common/BytesProcessorFactoryTests.java

@@ -0,0 +1,27 @@
+/*
+ * Licensed to Elasticsearch under one or more contributor
+ * license agreements. See the NOTICE file distributed with
+ * this work for additional information regarding copyright
+ * ownership. Elasticsearch licenses this file to you under
+ * the Apache License, Version 2.0 (the "License"); you may
+ * not use this file except in compliance with the License.
+ * You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing,
+ * software distributed under the License is distributed on an
+ * "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
+ * KIND, either express or implied. See the License for the
+ * specific language governing permissions and limitations
+ * under the License.
+ */
+
+package org.elasticsearch.ingest.common;
+
+public class BytesProcessorFactoryTests extends AbstractStringProcessorFactoryTestCase {
+ @Override
+ protected AbstractStringProcessor.Factory newFactory() {
+ return new BytesProcessor.Factory();
+ }
+}

modules/ingest-common/src/test/java/org/elasticsearch/ingest/common/BytesProcessorTests.java

@@ -0,0 +1,98 @@
+/*
+ * Licensed to Elasticsearch under one or more contributor
+ * license agreements. See the NOTICE file distributed with
+ * this work for additional information regarding copyright
+ * ownership. Elasticsearch licenses this file to you under
+ * the Apache License, Version 2.0 (the "License"); you may
+ * not use this file except in compliance with the License.
+ * You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing,
+ * software distributed under the License is distributed on an
+ * "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
+ * KIND, either express or implied. See the License for the
+ * specific language governing permissions and limitations
+ * under the License.
+ */
+
+package org.elasticsearch.ingest.common;
+
+import org.elasticsearch.ElasticsearchException;
+import org.elasticsearch.common.unit.ByteSizeUnit;
+import org.elasticsearch.common.unit.ByteSizeValue;
+import org.elasticsearch.ingest.IngestDocument;
+import org.elasticsearch.ingest.Processor;
+import org.elasticsearch.ingest.RandomDocumentPicks;
+import org.hamcrest.CoreMatchers;
+
+import static org.hamcrest.Matchers.equalTo;
+
+public class BytesProcessorTests extends AbstractStringProcessorTestCase {
+
+ private String modifiedInput;
+
+ @Override
+ protected AbstractStringProcessor newProcessor(String field, boolean ignoreMissing, String targetField) {
+ return new BytesProcessor(randomAlphaOfLength(10), field, ignoreMissing, targetField);
+ }
+
+ @Override
+ protected String modifyInput(String input) {
+ //largest value that allows all results < Long.MAX_VALUE bytes
+ long randomNumber = randomLongBetween(1, Long.MAX_VALUE / ByteSizeUnit.PB.toBytes(1));
+ ByteSizeUnit randomUnit = randomFrom(ByteSizeUnit.values());
+ modifiedInput = randomNumber + randomUnit.getSuffix();
+ return modifiedInput;
+ }
+
+ @Override
+ protected Long expectedResult(String input) {
+ return ByteSizeValue.parseBytesSizeValue(modifiedInput, null, "").getBytes();
+ }
+
+ @Override
+ protected Class<Long> expectedResultType() {
+ return Long.class;
+ }
+
+ public void testTooLarge() {
+ IngestDocument ingestDocument = RandomDocumentPicks.randomIngestDocument(random());
+ String fieldName = RandomDocumentPicks.addRandomField(random(), ingestDocument, "8912pb");
+ Processor processor = newProcessor(fieldName, randomBoolean(), fieldName);
+ ElasticsearchException exception = expectThrows(ElasticsearchException.class, () -> processor.execute(ingestDocument));
+ assertThat(exception.getMessage(),
+ CoreMatchers.equalTo("failed to parse setting [" + fieldName + "] with value [8912pb] as a size in bytes"));
+ assertThat(exception.getCause().getMessage(),
+ CoreMatchers.containsString("Values greater than 9223372036854775807 bytes are not supported"));
+ }
+
+ public void testNotBytes() {
+ IngestDocument ingestDocument = RandomDocumentPicks.randomIngestDocument(random());
+ String fieldName = RandomDocumentPicks.addRandomField(random(), ingestDocument, "junk");
+ Processor processor = newProcessor(fieldName, randomBoolean(), fieldName);
+ ElasticsearchException exception = expectThrows(ElasticsearchException.class, () -> processor.execute(ingestDocument));
+ assertThat(exception.getMessage(),
+ CoreMatchers.equalTo("failed to parse [junk]"));
+ }
+
+ public void testMissingUnits() {
+ IngestDocument ingestDocument = RandomDocumentPicks.randomIngestDocument(random());
+ String fieldName = RandomDocumentPicks.addRandomField(random(), ingestDocument, "1");
+ Processor processor = newProcessor(fieldName, randomBoolean(), fieldName);
+ ElasticsearchException exception = expectThrows(ElasticsearchException.class, () -> processor.execute(ingestDocument));
+ assertThat(exception.getMessage(),
+ CoreMatchers.containsString("unit is missing or unrecognized"));
+ }
+
+ public void testFractional() throws Exception {
+ IngestDocument ingestDocument = RandomDocumentPicks.randomIngestDocument(random());
+ String fieldName = RandomDocumentPicks.addRandomField(random(), ingestDocument, "1.1kb");
+ Processor processor = newProcessor(fieldName, randomBoolean(), fieldName);
+ processor.execute(ingestDocument);
+ assertThat(ingestDocument.getFieldValue(fieldName, expectedResultType()), equalTo(1126L));
+ assertWarnings("Fractional bytes values are deprecated. Use non-fractional bytes values instead: [1.1kb] found for setting " +
+ "[" + fieldName + "]");
+ }
+}

modules/ingest-common/src/test/resources/rest-api-spec/test/ingest/10_basic.yml

@@ -30,3 +30,4 @@
  - contains: { nodes.$master.ingest.processors: { type: split } }
  - contains: { nodes.$master.ingest.processors: { type: trim } }
  - contains: { nodes.$master.ingest.processors: { type: uppercase } }
+ - contains: { nodes.$master.ingest.processors: { type: bytes } }

modules/ingest-common/src/test/resources/rest-api-spec/test/ingest/180_bytes_processor.yml

@@ -0,0 +1,42 @@
+---
+teardown:
+ - do:
+ ingest.delete_pipeline:
+ id: "my_pipeline"
+ ignore: 404
+
+---
+"Test bytes processor":
+ - do:
+ ingest.put_pipeline:
+ id: "my_pipeline"
+ body: >
+ {
+ "description": "_description",
+ "processors": [
+ {
+ "bytes" : {
+ "field" : "bytes_source_field",
+ "target_field" : "bytes_target_field"
+ }
+ }
+ ]
+ }
+ - match: { acknowledged: true }
+
+ - do:
+ index:
+ index: test
+ type: test
+ id: 1
+ pipeline: "my_pipeline"
+ body: {bytes_source_field: "1kb"}
+
+ - do:
+ get:
+ index: test
+ type: test
+ id: 1
+ - match: { _source.bytes_source_field: "1kb" }
+ - match: { _source.bytes_target_field: 1024 }
+