Add truncate string processor

data-prepper-plugins/truncate-processor/README.md

@@ -0,0 +1,92 @@
+# Truncate Processor
+
+This is a processor that truncates key's value at the beginning or at the end or at both sides of a string as per the configuration. If the key's value is a list, then each of the string members of the list are truncated. Non-string members of the list are left untouched. If `truncate_when` option is provided, the truncation of the input is done only when the condition specified is true for the event being processed.
+
+## Basic Usage
+To get started, create the following `pipeline.yaml`.
+```yaml
+pipeline:
+  source:
+    file:
+      path: "/full/path/to/logs_json.log"
+      record_type: "event"
+      format: "json"
+  processor:
+    - trucate_string:
+        entries:
+          - source: "message"
+            length: 5
+  sink:
+    - stdout:
+```
+
+Create the following file named `logs_json.log` and replace the `path` in the file source of your `pipeline.yaml` with the path of this file.
+
+```json
+{"message": "hello,world"}
+```
+When you run Data Prepper with this `pipeline.yaml`, you should see the following output:
+
+```json
+{"message":["hello"]}
+```
+
+If the above yaml file has additional config of `start_at: 2`, then the output would be following:
+
+```json
+{"message":["llo,w"]}
+```
+
+If the above yaml file has additional config of `start_at: 2`, and does not have `length: 5` in the config, then the output would be following:
+
+```json
+{"message":["llo,world"]}
+```
+
+If the source has an list of strings, then the result will be an array of strings where each of the member of the list is truncated. The following input
+```json
+{"message": ["hello_one", "hello_two", "hello_three"]}
+```
+is transformed to the following:
+
+```json
+{"message": ["hello", "hello", "hello"]}
+```
+
+Example configuration with `truncate_when` option:
+```yaml
+pipeline:
+  source:
+    file:
+      path: "/full/path/to/logs_json.log"
+      record_type: "event"
+      format: "json"
+  processor:
+    - trucate_string:
+        entries:
+          - source: "message"
+            length: 5
+            start_at: 7
+            truncate_when: '/id == 1'
+  sink:
+    - stdout:
+```
+
+When the pipeline started with the above configuration receives the following two events
+```json
+{"message": "hello, world", "id": 1}
+{"message": "hello, world,not-truncated", "id": 2}
+```
+the output would be
+```json
+{"message": "world", "id": 1}
+{"message": "hello, world,not-truncated", "id": 2}
+```
+
+### Configuration
+* `entries` - (required) - A list of entries to add to an event
+    * `source` - (required) - The key to be modified
+    * `start_at` - (optional) - starting index of the string. Defaults to 0.
+    * `length` - (optional) - length of the string after truncation. Defaults to end of the string.
+Either `start_at` or `length` or both must be present
+

data-prepper-plugins/truncate-processor/build.gradle

@@ -0,0 +1,17 @@
+/*
+ * Copyright OpenSearch Contributors
+ * SPDX-License-Identifier: Apache-2.0
+ */
+
+plugins {
+    id 'java'
+}
+
+dependencies {
+    implementation project(':data-prepper-api')
+    implementation project(':data-prepper-test-common')
+    implementation project(':data-prepper-plugins:common')
+    implementation 'com.fasterxml.jackson.core:jackson-databind'
+    testImplementation libs.commons.lang3
+}
+

data-prepper-plugins/truncate-processor/src/main/java/org/opensearch/dataprepper/plugins/processor/truncate/TruncateStringProcessor.java

@@ -0,0 +1,94 @@
+/*
+ * Copyright OpenSearch Contributors
+ * SPDX-License-Identifier: Apache-2.0
+ */
+
+package org.opensearch.dataprepper.plugins.processor.truncate;
+
+import org.opensearch.dataprepper.expression.ExpressionEvaluator;
+import org.opensearch.dataprepper.metrics.PluginMetrics;
+import org.opensearch.dataprepper.model.annotations.DataPrepperPlugin;
+import org.opensearch.dataprepper.model.annotations.DataPrepperPluginConstructor;
+import org.opensearch.dataprepper.model.event.Event;
+import org.opensearch.dataprepper.model.record.Record;
+import org.opensearch.dataprepper.model.processor.AbstractProcessor;
+import org.opensearch.dataprepper.model.processor.Processor;
+
+import java.util.Collection;
+import java.util.ArrayList;
+import java.util.List;
+
+/**
+ * This processor takes in a key and truncates its value to a string with
+ * characters from the front or at the end or at both removed.
+ * If the value is not a string, no action is performed.
+ */
+@DataPrepperPlugin(name = "truncate", pluginType = Processor.class, pluginConfigurationType = TruncateStringProcessorConfig.class)
+public class TruncateStringProcessor extends AbstractProcessor<Record<Event>, Record<Event>>{
+    private final List<TruncateStringProcessorConfig.Entry> entries;
+    private final ExpressionEvaluator expressionEvaluator;
+
+    @DataPrepperPluginConstructor
+    public TruncateStringProcessor(final PluginMetrics pluginMetrics, final TruncateStringProcessorConfig config, final ExpressionEvaluator expressionEvaluator) {
+        super(pluginMetrics);
+        this.entries = config.getEntries();
+        this.expressionEvaluator = expressionEvaluator;
+    }
+
+    private String getTruncatedValue(final TruncateStringProcessorConfig.Entry entry, final String value) {
+        int startIndex = entry.getStartAt() == null ? 0 : entry.getStartAt();
+        Integer length = entry.getLength();
+        String truncatedValue = (length == null || startIndex+length >= value.length()) ? value.substring(startIndex) : value.substring(startIndex, startIndex + length);
+
+        return truncatedValue;
+    }
+
+    @Override
+    public Collection<Record<Event>> doExecute(final Collection<Record<Event>> records) {
+        for(final Record<Event> record : records) {
+            final Event recordEvent = record.getData();
+            for(TruncateStringProcessorConfig.Entry entry : entries) {
+                if (entry.getTruncateWhen() != null && !expressionEvaluator.evaluateConditional(entry.getTruncateWhen(), recordEvent)) {
+                    continue;
+                }
+                final String key = entry.getSource();
+                if (!recordEvent.containsKey(key)) {
+                    continue;
+                }
+
+                final Object value = recordEvent.get(key, Object.class);
+                if (value instanceof String) {
+                    recordEvent.put(key, getTruncatedValue(entry, (String)value));
+                } else if (value instanceof List) {
+                    List<Object> result = new ArrayList<>();
+                    for (Object arrayObject: (List)value) {
+                        if (arrayObject instanceof String) {
+                            result.add(getTruncatedValue(entry, (String)arrayObject));
+                        } else {
+                            result.add(arrayObject);
+                        }
+                    }
+                    recordEvent.put(key, result);
+                }
+            }
+        }
+
+        return records;
+    }
+
+    @Override
+    public void prepareForShutdown() {
+
+    }
+
+    @Override
+    public boolean isReadyForShutdown() {
+        return true;
+    }
+
+    @Override
+    public void shutdown() {
+
+    }
+}
+

data-prepper-plugins/truncate-processor/src/main/java/org/opensearch/dataprepper/plugins/processor/truncate/TruncateStringProcessorConfig.java

@@ -0,0 +1,76 @@
+/*
+ * Copyright OpenSearch Contributors
+ * SPDX-License-Identifier: Apache-2.0
+ */
+
+package org.opensearch.dataprepper.plugins.processor.truncate;
+
+import com.fasterxml.jackson.annotation.JsonProperty;
+import jakarta.validation.Valid;
+import jakarta.validation.constraints.NotEmpty;
+import jakarta.validation.constraints.NotNull;
+import jakarta.validation.constraints.AssertTrue; 
+
+import java.util.List;
+
+public class TruncateStringProcessorConfig {
+    public static class Entry {
+
+        @NotEmpty
+        @NotNull
+        private String source;
+
+        @JsonProperty("length")
+        private Integer length;
+
+        @JsonProperty("start_at")
+        private Integer startAt;
+
+        @JsonProperty("truncate_when")
+        private String truncateWhen;
+
+        public String getSource() {
+            return source;
+        }
+
+        public Integer getStartAt() {
+            return startAt;
+        }
+
+        public Integer getLength() {
+            return length;
+        }
+
+        @AssertTrue(message = "At least one of start_at or length or both must be specified and the values must be positive integers")
+        public boolean hasStartAtOrLength() {
+            if (length == null && startAt == null) {
+                return false;
+            }
+            if (length != null && length < 0) {
+                return false;
+            }
+            if (startAt != null && startAt < 0) {
+                return false;
+            }
+            return true;
+        }  
+
+        public String getTruncateWhen() { return truncateWhen; }
+
+        public Entry(final String source, final Integer startAt, final Integer length, final String truncateWhen) {
+            this.source = source;
+            this.startAt = startAt;
+            this.length = length;
+            this.truncateWhen = truncateWhen;
+        }
+
+        public Entry() {}
+    }
+
+    private List<@Valid Entry> entries;
+
+    public List<Entry> getEntries() {
+        return entries;
+    }
+}
+