Add filter_list processor to filter list elements

Signed-off-by: Manisha Yadav <yavmanis@amazon.com>

Author: yavmanis

data-prepper-plugins/mutate-event-processors/README.md

@@ -647,6 +647,153 @@ will end up with this after processing:
 * `tags_on_failure` - (optional): a list of tags to add to event metadata when the event fails to process
 
 
+## FilterListProcessor
+A processor that filters elements within an array field by evaluating a condition against each element, keeping only those where the condition is true. It supports arrays of objects and arrays of primitives (strings, numbers, booleans).
+
+### 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:
+    - filter_list:
+        source: "items"
+        keep_when: '/status == "active"'
+  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
+{"items": [{"name": "item1", "status": "active"}, {"name": "item2", "status": "inactive"}, {"name": "item3", "status": "active"}]}
+```
+
+When run, the processor will filter the array in-place and produce the following output:
+
+```json
+{"items": [{"name": "item1", "status": "active"}, {"name": "item3", "status": "active"}]}
+```
+
+### Filtering to a different target
+
+You can write the filtered result to a different key, leaving the original array unchanged:
+
+```yaml
+  processor:
+    - filter_list:
+        source: "items"
+        target: "active_items"
+        keep_when: '/status == "active"'
+```
+
+With the same input, the output will be:
+
+```json
+{
+  "items": [{"name": "item1", "status": "active"}, {"name": "item2", "status": "inactive"}, {"name": "item3", "status": "active"}],
+  "active_items": [{"name": "item1", "status": "active"}, {"name": "item3", "status": "active"}]
+}
+```
+
+### Filtering primitive arrays
+
+For arrays of primitives (strings, numbers, booleans), each element is accessible via the `/value` key in the expression:
+
+```yaml
+  processor:
+    - filter_list:
+        source: "tags"
+        keep_when: '/value != ""'
+```
+
+With the following input:
+
+```json
+{"tags": ["important", "", "urgent", ""]}
+```
+
+The output will be:
+
+```json
+{"tags": ["important", "urgent"]}
+```
+
+Another example filtering numbers:
+
+```yaml
+  processor:
+    - filter_list:
+        source: "scores"
+        keep_when: '/value > 50'
+```
+
+With the following input:
+
+```json
+{"scores": [90, 30, 75, 10]}
+```
+
+The output will be:
+
+```json
+{"scores": [90, 75]}
+```
+
+### Using both conditions
+
+The `filter_list_when` condition controls whether the processor runs at all (evaluated against the root event), while `keep_when` controls which elements are kept (evaluated per element):
+
+```yaml
+  processor:
+    - filter_list:
+        source: "items"
+        keep_when: '/status == "active"'
+        filter_list_when: '/env == "production"'
+```
+
+With the following input:
+
+```json
+{"env": "production", "items": [{"name": "item1", "status": "active"}, {"name": "item2", "status": "inactive"}]}
+```
+
+Since `env` is `"production"`, the processor runs and filters by `status`, producing:
+
+```json
+{"env": "production", "items": [{"name": "item1", "status": "active"}]}
+```
+
+With a different event where `filter_list_when` evaluates to `false`:
+
+```json
+{"env": "staging", "items": [{"name": "item1", "status": "active"}, {"name": "item2", "status": "inactive"}]}
+```
+
+The processor is skipped entirely and the event passes through unchanged:
+
+```json
+{"env": "staging", "items": [{"name": "item1", "status": "active"}, {"name": "item2", "status": "inactive"}]}
+```
+
+### Configuration
+* `source` - (required) - The key of the array field to filter. Supports nested paths (e.g. `outer_key/inner_list`).
+* `target` - (optional) - The key to write the filtered array to. Defaults to the `source` key (in-place filtering). Supports nested paths.
+* `keep_when` - (required) - A [Data Prepper expression](https://opensearch.org/docs/latest/data-prepper/pipelines/expression-syntax/) evaluated per element. Elements where this expression evaluates to `true` are kept. For object elements, the expression is evaluated against the object's fields directly (e.g. `/status == "active"`). For primitive elements, the value is accessible via `/value` (e.g. `/value > 50`). When no elements match, the result is an empty list `[]`.
+* `filter_list_when` - (optional) - A [Data Prepper expression](https://opensearch.org/docs/latest/data-prepper/pipelines/expression-syntax/) evaluated against the root event. When provided, the processor only runs if this condition is `true`. By default, all events are processed.
+* `tags_on_failure` - (optional) - A list of tags to add to the event metadata when the processor fails to process the event.
+
+**Edge case behavior:**
+- If the `source` key does not exist or its value is `null`, the processor is a no-op and the event passes through unchanged.
+- If the `source` value is not a list (e.g. a string or number), the processor logs a warning and adds `tags_on_failure` if configured.
+- `null` elements within the list are evaluated normally. For example, with `keep_when: '/value != null'`, null elements are filtered out while non-null elements are kept.
+
+___
+
 ## Developer Guide
 This plugin is compatible with Java 11 and 17. Refer to the following developer guides for plugin development:
 - [Developer Guide](https://github.com/opensearch-project/data-prepper/blob/main/docs/developer_guide.md)

data-prepper-plugins/mutate-event-processors/src/main/java/org/opensearch/dataprepper/plugins/processor/mutateevent/FilterListProcessor.java

@@ -0,0 +1,149 @@
+/*
+ * Copyright OpenSearch Contributors
+ * SPDX-License-Identifier: Apache-2.0
+ *
+ * The OpenSearch Contributors require contributions made to
+ * this file be licensed under the Apache-2.0 license or a
+ * compatible open source license.
+ */
+
+package org.opensearch.dataprepper.plugins.processor.mutateevent;
+
+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.event.JacksonEvent;
+import org.opensearch.dataprepper.model.plugin.InvalidPluginConfigurationException;
+import org.opensearch.dataprepper.model.processor.AbstractProcessor;
+import org.opensearch.dataprepper.model.processor.Processor;
+import org.opensearch.dataprepper.model.record.Record;
+import org.slf4j.Logger;
+import org.slf4j.LoggerFactory;
+
+import java.util.ArrayList;
+import java.util.Collection;
+import java.util.Collections;
+import java.util.List;
+import java.util.Map;
+import java.util.Objects;
+
+import static org.opensearch.dataprepper.logging.DataPrepperMarkers.EVENT;
+import static org.opensearch.dataprepper.logging.DataPrepperMarkers.NOISY;
+
+@DataPrepperPlugin(name = "filter_list", pluginType = Processor.class, pluginConfigurationType = FilterListProcessorConfig.class)
+public class FilterListProcessor extends AbstractProcessor<Record<Event>, Record<Event>> {
+
+    private static final Logger LOG = LoggerFactory.getLogger(FilterListProcessor.class);
+    private final FilterListProcessorConfig config;
+    private final ExpressionEvaluator expressionEvaluator;
+    private final String target;
+
+    @DataPrepperPluginConstructor
+    public FilterListProcessor(final PluginMetrics pluginMetrics, final FilterListProcessorConfig config, final ExpressionEvaluator expressionEvaluator) {
+        super(pluginMetrics);
+        this.config = config;
+        this.expressionEvaluator = expressionEvaluator;
+        this.target = config.getTarget() != null ? config.getTarget() : config.getSource();
+
+        if (config.getFilterListWhen() != null
+                && !expressionEvaluator.isValidExpressionStatement(config.getFilterListWhen())) {
+            throw new InvalidPluginConfigurationException(
+                    String.format("filter_list_when %s is not a valid expression statement. " +
+                                    "See https://opensearch.org/docs/latest/data-prepper/pipelines/expression-syntax/ for valid expression syntax",
+                            config.getFilterListWhen()));
+        }
+
+        if (!expressionEvaluator.isValidExpressionStatement(config.getKeepWhen())) {
+            throw new InvalidPluginConfigurationException(
+                    String.format("keep_when %s is not a valid expression statement. " +
+                                    "See https://opensearch.org/docs/latest/data-prepper/pipelines/expression-syntax/ for valid expression syntax",
+                            config.getKeepWhen()));
+        }
+    }
+
+    @Override
+    public Collection<Record<Event>> doExecute(final Collection<Record<Event>> records) {
+        for (final Record<Event> record : records) {
+            final Event recordEvent = record.getData();
+
+            try {
+                if (Objects.nonNull(config.getFilterListWhen()) && !expressionEvaluator.evaluateConditional(config.getFilterListWhen(), recordEvent)) {
+                    continue;
+                }
+
+                final List<Object> sourceList;
+                try {
+                    sourceList = recordEvent.get(config.getSource(), List.class);
+                } catch (final Exception e) {
+                    LOG.warn(EVENT, "Given source path [{}] is not valid on record [{}]",
+                            config.getSource(), recordEvent, e);
+                    addTagsOnFailure(recordEvent);
+                    continue;
+                }
+
+                if (sourceList == null) {
+                    LOG.debug("Source list at path [{}] is null, skipping event", config.getSource());
+                    continue;
+                }
+
+                final List<Object> filteredList = new ArrayList<>();
+                final JacksonEvent.Builder contextBuilder = JacksonEvent.builder()
+                        .withEventType("event");
+
+                for (final Object element : sourceList) {
+                    @SuppressWarnings("unchecked")
+                    final Map<String, Object> contextMap = element instanceof Map
+                            ? (Map<String, Object>) element
+                            : Collections.singletonMap("value", element);
+
+                    try {
+                        final Event elementEvent = contextBuilder
+                                .withData(contextMap)
+                                .build();
+
+                        if (expressionEvaluator.evaluateConditional(config.getKeepWhen(), elementEvent)) {
+                            filteredList.add(element);
+                        }
+                    } catch (final Exception e) {
+                        LOG.warn(EVENT, "Error evaluating keep_when expression [{}] for element in source list at path [{}]",
+                                config.getKeepWhen(), config.getSource(), e);
+                    }
+                }
+
+                recordEvent.put(target, filteredList);
+
+            } catch (final Exception e) {
+                LOG.atError()
+                        .addMarker(EVENT)
+                        .addMarker(NOISY)
+                        .setMessage("There was an exception while processing Event [{}]")
+                        .addArgument(recordEvent)
+                        .setCause(e)
+                        .log();
+                addTagsOnFailure(recordEvent);
+            }
+        }
+        return records;
+    }
+
+    private void addTagsOnFailure(final Event event) {
+        if (config.getTagsOnFailure() != null) {
+            event.getMetadata().addTags(config.getTagsOnFailure());
+        }
+    }
+
+    @Override
+    public void prepareForShutdown() {
+    }
+
+    @Override
+    public boolean isReadyForShutdown() {
+        return true;
+    }
+
+    @Override
+    public void shutdown() {
+    }
+}

data-prepper-plugins/mutate-event-processors/src/main/java/org/opensearch/dataprepper/plugins/processor/mutateevent/FilterListProcessorConfig.java

@@ -0,0 +1,89 @@
+/*
+ * Copyright OpenSearch Contributors
+ * SPDX-License-Identifier: Apache-2.0
+ *
+ * The OpenSearch Contributors require contributions made to
+ * this file be licensed under the Apache-2.0 license or a
+ * compatible open source license.
+ */
+
+package org.opensearch.dataprepper.plugins.processor.mutateevent;
+
+import com.fasterxml.jackson.annotation.JsonClassDescription;
+import com.fasterxml.jackson.annotation.JsonProperty;
+import com.fasterxml.jackson.annotation.JsonPropertyDescription;
+import com.fasterxml.jackson.annotation.JsonPropertyOrder;
+import jakarta.validation.constraints.NotEmpty;
+import jakarta.validation.constraints.NotNull;
+import org.opensearch.dataprepper.model.annotations.ExampleValues;
+import org.opensearch.dataprepper.model.annotations.ExampleValues.Example;
+
+import java.util.List;
+
+@JsonPropertyOrder
+@JsonClassDescription("The <code>filter_list</code> processor evaluates a condition against each element of an array " +
+        "and keeps only those elements where the condition is true.")
+public class FilterListProcessorConfig {
+
+    @NotNull
+    @NotEmpty
+    @JsonProperty("source")
+    @JsonPropertyDescription("The key of the array field to filter. Supports nested paths.")
+    @ExampleValues({
+            @Example(value = "my-list", description = "Filters the 'my-list' array at the root of the event."),
+            @Example(value = "outer-key/my-list", description = "Filters the 'my-list' array nested under 'outer-key'.")
+    })
+    private String source;
+
+    @JsonProperty("target")
+    @JsonPropertyDescription("The key to write the filtered array to. Defaults to the source key (in-place). " +
+            "Supports nested paths — intermediate objects are created automatically if they do not exist.")
+    @ExampleValues({
+            @Example(value = "filtered-list", description = "Writes the filtered array to 'filtered-list'.")
+    })
+    private String target;
+
+    @NotNull
+    @NotEmpty
+    @JsonProperty("keep_when")
+    @JsonPropertyDescription("An expression evaluated per element. Elements where this expression evaluates to true are kept. " +
+            "The expression is evaluated against each element of the array as if it were a standalone event.")
+    @ExampleValues({
+            @Example(value = "/status == \"active\"", description = "Keeps only elements where 'status' equals 'active'."),
+            @Example(value = "/score > 50", description = "Keeps only elements where 'score' is greater than 50.")
+    })
+    private String keepWhen;
+
+    @JsonProperty("filter_list_when")
+    @JsonPropertyDescription("A <a href=\"https://opensearch.org/docs/latest/data-prepper/pipelines/expression-syntax/\">conditional expression</a>, " +
+            "such as <code>/some-key == \"test\"</code>, that will be evaluated against the root event to determine whether " +
+            "the processor will be run on the event. By default, all events will be processed unless otherwise stated.")
+    @ExampleValues({
+            @Example(value = "/some-key == \"test\"", description = "The processor only runs when the value of 'some-key' is 'test'.")
+    })
+    private String filterListWhen;
+
+    @JsonProperty("tags_on_failure")
+    @JsonPropertyDescription("A list of tags to add to the event metadata when the event fails to process.")
+    private List<String> tagsOnFailure;
+
+    public String getSource() {
+        return source;
+    }
+
+    public String getTarget() {
+        return target;
+    }
+
+    public String getKeepWhen() {
+        return keepWhen;
+    }
+
+    public String getFilterListWhen() {
+        return filterListWhen;
+    }
+
+    public List<String> getTagsOnFailure() {
+        return tagsOnFailure;
+    }
+}