[Streams] Update processors for 9.3 release (#4614)

Author: mdbirnstiehl

solutions/observability/streams/management/extract.md

@@ -3,9 +3,9 @@ applies_to:
   serverless: ga
   stack: preview =9.1, ga 9.2+
 ---
-# Extract fields [streams-extract-fields]
+# Process documents [streams-extract-fields]
 
-After selecting a stream, use the **Processing** tab to add [processors](#streams-extract-processors) that extract meaningful fields from your log messages. These fields let you filter and analyze your data more effectively.
+After selecting a stream, use the **Processing** tab to add [processors](#streams-extract-processors) and [conditions](#streams-add-processor-conditions) that modify your documents and extract meaningful fields, so you can filter and analyze your data more effectively.
 
 For example, in [Discover](../../../../explore-analyze/discover.md), extracted fields might let you filter for log messages with an `ERROR` log level that occurred during a specific time period to help diagnose an issue. Without extracting the log level and timestamp fields from your messages, those filters wouldn't return meaningful results.
 
@@ -14,7 +14,7 @@ The **Processing** tab also:
 - Simulates your processors and provides an immediate [preview](#streams-preview-changes) that's tested end to end
 - Flags indexing issues, like [mapping conflicts](#streams-processing-mapping-conflicts), so you can address them before applying changes
 
-After creating your processor, all future data ingested into the stream is parsed into structured fields accordingly.
+After creating your processor, Streams parses all future data ingested into the stream into structured fields accordingly.
 
 :::{note}
 Applied changes aren't retroactive and only affect *future ingested data*.
@@ -24,13 +24,28 @@ Applied changes aren't retroactive and only affect *future ingested data*.
 
 Streams supports the following processors:
 
+- [**Drop**](./extract/drop.md): Drops the document without raising any errors. This is useful to prevent the document from getting indexed based on a condition.
+- [**Remove**](./extract/remove.md): Removes existing fields.
 - [**Date**](./extract/date.md): Converts date strings into timestamps, with options for timezone, locale, and output formatting.
+- [**Convert**](./extract/convert.md): Converts a field in the currently ingested document to a different type, such as converting a string to an integer.
+- [**Replace**](./extract/replace.md): Replaces parts of a string field according to a regular expression pattern with a replacement string.
 - [**Dissect**](./extract/dissect.md): Extracts fields from structured log messages using defined delimiters instead of patterns, making it faster than Grok and ideal for consistently formatted logs.
 - [**Grok**](./extract/grok.md): Extracts fields from unstructured log messages using predefined or custom patterns, supports multiple match attempts in sequence, and can automatically generate patterns with an [LLM connector](/explore-analyze/ai-features/llm-guides/llm-connectors.md).
 - [**Set**](./extract/set.md): Assigns a specific value to a field, creating the field if it doesn’t exist or overwriting its value if it does.
+- [**Math**](./extract/math.md): Evaluates arithmetic or logical expressions.
 - [**Rename**](./extract/rename.md): Changes the name of a field, moving its value to a new field name and removing the original.
 - [**Append**](./extract/append.md): Adds a value to an existing array field, or creates the field as an array if it doesn’t exist.
 
+### Processor limitations and inconsistencies [streams-processor-inconsistencies]
+
+Streams exposes a Streamlang configuration, but internally it relies on {{es}} ingest pipeline processors and ES|QL. Streamlang doesn’t always have 1:1 parity with the ingest processors because it needs to support options that work in both ingest pipelines and ES|QL. In most cases, you won’t need to worry about these details, but the underlying design decisions still affect the UI and available configuration options. The following are some limitations and inconsistencies when using Streamlang processors:
+
+- **Consistently typed fields**: ES|QL requires one consistent type per column, so workflows that produce mixed types across documents won’t transpile.
+- **Conversion of types**: ES|QL and ingest pipelines accept different conversion combinations and strictness (especially for strings), so `convert` can behave differently across targets.
+- **Multi-value commands/functions**: Fields can contain one or multiple values. ES|QL and ingest processors don’t always handle these cases the same way. For example, grok in ES|QL handles multiple values automatically, while the grok processor does not
+- **Conditional execution**: ES|QL's enforced table shape limits conditional casting, parsing, and wildcard field operations that ingest pipelines can do per-document.
+- **Arrays of objects / flattening**: Ingest pipelines preserve nested JSON arrays, while ES|QL flattens to columns, so operations like rename and delete on parent objects can differ or fail.
+
 ## Add a processor [streams-add-processors]
 
 Streams uses [{{es}} ingest pipelines](../../../../manage-data/ingest/transform-enrich/ingest-pipelines.md) made up of processors to transform your data, without requiring you to switch interfaces and manually update pipelines.
@@ -49,7 +64,7 @@ Refer to individual [supported processors](#streams-extract-processors) for more
 Editing processors with JSON is planned for a future release, and additional processors may be supported over time.
 :::
 
-### Add conditions to processors [streams-add-processor-conditions]
+### Add conditions [streams-add-processor-conditions]
 
 You can add conditions to processors so they only run on data that meets those conditions. Each condition is a boolean expression that's evaluated for every document.
 
@@ -76,6 +91,8 @@ Streams processors support the following comparators:
 - not exists
 :::
 
+After creating a condition, add a processor or another condition to it by selecting the {icon}`plus_in_circle`.
+
 ### Preview changes [streams-preview-changes]
 
 After you create processors, the **Data preview** tab simulates processor results with additional filtering options depending on the outcome of the simulation.
@@ -93,9 +110,9 @@ After making sure everything in the **Data preview** tab is correct, select **Sa
 
 If you edit the stream after saving your changes, keep the following in mind:
 
-- Adding processors to the end of the list will work as expected.
-- Editing or reordering existing processors can cause inaccurate results. Because the pipeline may have already processed the documents used for sampling, **Data preview** cannot accurately simulate changes to existing data.
-- Adding a new processor and moving it before an existing processor may cause inaccurate results. **Data preview** only simulates the new processor, not the existing ones, so the simulation may not accurately reflect changes to existing data.
+- Adding processors to the end of the list works as expected.
+- Editing or reordering existing processors can cause inaccurate results. Because the pipeline might have already processed the documents used for sampling, **Data preview** cannot accurately simulate changes to existing data.
+- Adding a new processor and moving it before an existing processor can cause inaccurate results. **Data preview** only simulates the new processor, not the existing ones, so the simulation may not accurately reflect changes to existing data.
 
 ### Ignore failures [streams-ignore-failures]
 
@@ -122,7 +139,7 @@ Selecting **Failed** shows the documents that weren't parsed correctly:
 :screenshot:
 :::
 
-Failures are displayed at the bottom of the process editor. Some failures may require fixes, while others simply serve as a warning:
+Streams displays failures at the bottom of the process editor. Some failures might require fixes, while others serve as a warning:
 
 :::{image} ../../../images/logs-streams-processor-failures.png
 :screenshot:
@@ -179,10 +196,10 @@ Streams then creates and manages the `<data_stream_name>@stream.processing` pipe
 ### User interaction with pipelines
 
 Do not manually modify the `<data_stream_name>@stream.processing` pipeline created by Streams.
-You can still add your own processors manually to the `@custom` pipeline if needed. Adding processors before the pipeline processor created by Streams may cause unexpected behavior.
+You can still add your own processors manually to the `@custom` pipeline if needed. Adding processors before the pipeline processor created by Streams might cause unexpected behavior.
 
 ## Known limitations [streams-known-limitations]
 
 - Streams does not support all processors. More processors will be added in future versions.
-- The data preview simulation may not accurately reflect the changes to the existing data when editing existing processors or re-ordering them. Streams will allow proper simulations using original documents in a future version.
+- The data preview simulation might not accurately reflect the changes to the existing data when editing existing processors or re-ordering them. Streams will allow proper simulations using original documents in a future version.
 - Streams can't properly handle arrays. While it supports basic actions like appending or renaming, it can't access individual array elements. For classic streams, the workaround is to use the [manual pipeline configuration](./extract/manual-pipeline-configuration.md) that supports Painless scripting and all ingest processors.

solutions/observability/streams/management/extract/append.md

@@ -6,7 +6,7 @@ applies_to:
 # Append processor [streams-append-processor]
 % Need use cases
 
-Use the append processor to add a value to an existing array field, or create the field as an array if it doesn’t exist.
+Use the **Append** processor to add a value to an existing array field, or create the field as an array if it doesn’t exist.
 
 To use an append processor:
 
@@ -15,4 +15,4 @@ To use an append processor:
 1. Set **Source Field** to the field you want append values to.
 1. Set **Target field** to the values you want to append to the **Source Field**.
 
-This functionality uses the {{es}} rename pipeline processor. Refer to the [rename processor](elasticsearch://reference/enrich-processor/rename-processor.md) {{es}} documentation for more information.
+This functionality uses the {{es}} [append processor](elasticsearch://reference/enrich-processor/append-processor.md) internally, but you configure it in Streamlang. Streamlang doesn’t always have 1:1 parity with the ingest processor options and behavior. Refer to [Processor limitations and inconsistencies](../extract.md#streams-processor-inconsistencies).

solutions/observability/streams/management/extract/convert.md

@@ -0,0 +1,22 @@
+---
+applies_to:
+  serverless: ga
+  stack: ga 9.3+
+---
+
+# Convert processor [streams-convert-processor]
+The **Convert** processor converts a field to a different data type. For example, you could convert a string to an integer.
+
+To convert a field to a different data type:
+
+1. Select **Create** → **Create processor**.
+1. Select **Convert** from the **Processor** menu.
+1. Set the **Source Field** to the field you want to convert.
+1. (Optional) Set **Target field** to write the converted value to a different field.
+1. Set **Type** to the output data type.
+
+::::{note}
+If you add a **Convert** processor inside a condition group (a **WHERE** block), you must set a **Target field**.
+::::
+
+This functionality uses the {{es}} [Convert processor](elasticsearch://reference/enrich-processor/convert-processor.md) internally, but you configure it in Streamlang. Streamlang doesn’t always have 1:1 parity with the ingest processor options and behavior. Refer to [Processor limitations and inconsistencies](../extract.md#streams-processor-inconsistencies).

solutions/observability/streams/management/extract/date.md

@@ -6,7 +6,7 @@ applies_to:
 
 # Date processor [streams-date-processor]
 
-The date processor parses dates from fields, and then uses the date or timestamp as the timestamp for the document.
+The **Date** processor parses dates from fields, and then uses the date or timestamp as the timestamp for the document.
 
 To extract a timestamp field using the date processor:
 
@@ -15,7 +15,7 @@ To extract a timestamp field using the date processor:
 1. Set the **Source Field** to the field containing the timestamp.
 1. Set the **Format** field to one of the accepted date formats (ISO8602, UNIX, UNIX_MS, or TAI64N) or use a Java time pattern. Refer to the [example formats](#streams-date-examples) for more information.
 
-This functionality uses the {{es}} date pipeline processor. Refer to the [date processor](elasticsearch://reference/enrich-processor/date-processor.md) {{es}} documentation for more information.
+This functionality uses the {{es}} [Date processor](elasticsearch://reference/enrich-processor/date-processor.md) internally, but you configure it in Streamlang. Streamlang doesn’t always have 1:1 parity with the ingest processor options and behavior. Refer to [Processor limitations and inconsistencies](../extract.md#streams-processor-inconsistencies).
 
 ## Example formats [streams-date-examples]
 

solutions/observability/streams/management/extract/dissect.md

@@ -5,18 +5,18 @@ applies_to:
 ---
 # Dissect processor [streams-dissect-processor]
 
-The dissect processor parses structured log messages and extracts fields from them. It uses a set of delimiters to split the log message into fields instead of predefined patterns to match the log messages.
+The **Dissect** processor parses structured log messages and extracts fields from them. It uses a set of delimiters to split the log message into fields instead of predefined patterns to match the log messages.
 
 Dissect is much faster than Grok, and is recommend for log messages that follow a consistent, structured format.
 
 To parse a log message with a dissect processor:
 
 1. Select **Create** → **Create processor**.
 1. Select **Dissect** from the **Processor** menu.
-1. Set the **Source Field** to the field you want to dissect
+1. Set the **Source Field** to the field you want to dissect.
 1. Set the delimiters you want to use in the **Pattern** field. Refer to the [example pattern](#streams-dissect-example) for more information on setting delimiters.
 
-This functionality uses the {{es}} dissect pipeline processor. Refer to the [dissect processor](elasticsearch://reference/enrich-processor/dissect-processor.md) {{es}} documentation for more information.
+This functionality uses the {{es}} [Dissect processor](elasticsearch://reference/enrich-processor/dissect-processor.md) internally, but you configure it in Streamlang. Streamlang doesn’t always have 1:1 parity with the ingest processor options and behavior. Refer to [Processor limitations and inconsistencies](../extract.md#streams-processor-inconsistencies).
 
 ## Example dissect pattern [streams-dissect-example]
 

solutions/observability/streams/management/extract/drop.md

@@ -0,0 +1,21 @@
+---
+applies_to:
+  serverless: ga
+  stack: ga 9.3+
+---
+
+# Drop document processor [streams-drop-processor]
+
+The **Drop document** processor prevents documents from being indexed when they meet a specific condition, without raising an error.
+
+To configure a condition for dropping documents:
+
+1. Select **Create** → **Create processor**.
+1. Select **Drop document** from the **Processor** menu.
+1. Set the **Condition** for when you want to drop a document.
+
+  :::{warning}
+  The default is the `always` condition. Not setting a specific condition results in every document that matches the drop condition getting dropped from indexing.
+  :::
+
+This functionality uses the {{es}} [Drop processor](elasticsearch://reference/enrich-processor/drop-processor.md) internally, but you configure it in Streamlang. Streamlang doesn’t always have 1:1 parity with the ingest processor options and behavior. Refer to [Processor limitations and inconsistencies](../extract.md#streams-processor-inconsistencies).

solutions/observability/streams/management/extract/grok.md

@@ -5,7 +5,7 @@ applies_to:
 ---
 # Grok processor [streams-grok-processor]
 
-The grok processor parses unstructured log messages using a set of predefined patterns to match the log messages and extract the fields. The grok processor is very powerful and can parse a wide variety of log formats.
+The **Grok** processor parses unstructured log messages using a set of predefined patterns to match the log messages and extract the fields. The grok processor is powerful and can parse a wide variety of log formats.
 
 You can provide multiple patterns to the grok processor. The grok processor tries to match the log message against each pattern in the order they are provided. If a pattern matches, it extracts the fields and the remaining patterns won't be used.
 
@@ -20,7 +20,7 @@ To parse a log message with a grok processor:
 1. Set the **Source Field** to the field you want to search for grok matches.
 1. Set the patterns you want to use in the **Grok patterns** field. Refer to the [example pattern](#streams-grok-example) for more information on patterns.
 
-This functionality uses the {{es}} Grok pipeline processor. Refer to the [Grok processor](elasticsearch://reference/enrich-processor/grok-processor.md) {{es}} documentation for more information.
+This functionality uses the {{es}} [Grok processor](elasticsearch://reference/enrich-processor/grok-processor.md) internally, but you configure it in Streamlang. Streamlang doesn’t always have 1:1 parity with the ingest processor options and behavior. Refer to [Processor limitations and inconsistencies](../extract.md#streams-processor-inconsistencies).
 
 ## Example grok pattern [streams-grok-example]
 

solutions/observability/streams/management/extract/manual-pipeline-configuration.md

@@ -6,7 +6,7 @@ applies_to:
 # Manual pipeline configuration [streams-manual-pipeline-configuration]
 
 :::{note}
-The manual pipeline configuration processor is only available on [classic streams](../../streams.md#streams-classic-vs-wired).
+The **manual pipeline configuration** processor is only available on [classic streams](../../streams.md#streams-classic-vs-wired).
 :::
 
 The **Manual pipeline configuration** lets you create a JSON-encoded array of ingest pipeline processors.This is helpful if you want to add more advanced processing that isn't currently available as part of the UI-based processors.

solutions/observability/streams/management/extract/math.md

@@ -0,0 +1,16 @@
+---
+applies_to:
+  serverless: ga
+  stack: ga 9.3+
+---
+
+# Math processor [streams-math-processor]
+
+The **Math** processor evaluates arithmetic or logical expressions and stores the result in the target field.
+
+To calculate a value using an expression and store the result in a target field:
+
+1. Select **Create** → **Create processor**.
+1. Select **Math** from the **Processor** menu.
+1. Set the **Target field** where you want to write the expression result.
+1. Set your expression in the **Expression** field. You can directly reference fields in your expression (for example, `bytes / duration`).

solutions/observability/streams/management/extract/remove.md

@@ -0,0 +1,17 @@
+---
+applies_to:
+  serverless: ga
+  stack: ga 9.3+
+---
+
+# Remove processor [streams-remove-processor]
+
+The **Remove** processor removes a field (**Remove**) or removes a field and all its nested fields (**Remove by prefix**) from your documents.
+
+To remove a field:
+
+1. Select **Create** → **Create processor**.
+1. From the **Processor** menu, select **Remove** to remove a field or **Remove by prefix** to remove a field and all its nested fields.
+1. Set the **Source Field** to the field you want to remove.
+
+This functionality uses the {{es}} [Remove processor](elasticsearch://reference/enrich-processor/remove-processor.md) internally, but you configure it in Streamlang. Streamlang doesn’t always have 1:1 parity with the ingest processor options and behavior. Refer to [Processor limitations and inconsistencies](../extract.md#streams-processor-inconsistencies).