[8.x] [Streams 🌊] Stream enrichment processors management (#204793) (#206255)

# Backport This will backport the following commits from `main` to `8.x`: - [[Streams 🌊] Stream enrichment processors management (#204793)](https://github.com/elastic/kibana/pull/204793)  ### Questions ? Please refer to the [Backport tool documentation](https://github.com/sqren/backport)  --------- Co-authored-by: kibanamachine <42973632+kibanamachine@users.noreply.github.com>
2025-04-23 17:28:26 -04:00 · 2025-01-13 12:16:02 +01:00 · 2025-01-13 12:16:02 +01:00 · 432643af89
commit 432643af89
parent cdcb54fcbc
74 changed files with 3000 additions and 115 deletions
--- a/.github/CODEOWNERS
+++ b/.github/CODEOWNERS
@ -659,6 +659,7 @@ src/platform/plugins/shared/newsfeed @elastic/kibana-core
 test/common/plugins/newsfeed @elastic/kibana-core
 src/platform/plugins/private/no_data_page @elastic/appex-sharedux
 x-pack/platform/plugins/shared/notifications @elastic/appex-sharedux
+src/platform/packages/shared/kbn-object-utils @elastic/kibana-core
 src/platform/packages/shared/kbn-object-versioning @elastic/appex-sharedux
 src/platform/packages/shared/kbn-object-versioning-utils @elastic/appex-sharedux
 x-pack/solutions/observability/plugins/observability_ai_assistant_app @elastic/obs-ai-assistant
--- a/package.json
+++ b/package.json
@ -688,6 +688,7 @@
    "@kbn/newsfeed-test-plugin": "link:test/common/plugins/newsfeed",
    "@kbn/no-data-page-plugin": "link:src/platform/plugins/private/no_data_page",
    "@kbn/notifications-plugin": "link:x-pack/platform/plugins/shared/notifications",
+    "@kbn/object-utils": "link:src/platform/packages/shared/kbn-object-utils",
    "@kbn/object-versioning": "link:src/platform/packages/shared/kbn-object-versioning",
    "@kbn/object-versioning-utils": "link:src/platform/packages/shared/kbn-object-versioning-utils",
    "@kbn/observability-ai-assistant-app-plugin": "link:x-pack/solutions/observability/plugins/observability_ai_assistant_app",
--- a/src/platform/packages/shared/kbn-object-utils/README.md
+++ b/src/platform/packages/shared/kbn-object-utils/README.md
@ -0,0 +1,76 @@
+# @kbn/object-utils
+
+Utilities for objects manipulation and parsing.
+
+## Utilities
+
+### calculateObjectDiff
+
+This utils compares two JSON objects and calculates the added and removed properties, including nested properties.
+
+```ts
+const oldObject = {
+  alpha: 1,
+  beta: {
+    gamma: 2,
+    delta: {
+      sigma: 7,
+    },
+  },
+};
+
+const newObject = {
+  alpha: 1,
+  beta: {
+    gamma: 2,
+    eta: 4,
+  },
+};
+
+const diff = calculateObjectDiff(oldObject, newObject);
+
+/*
+Result:
+{
+  added: {
+    beta: {
+      eta: 4,
+    },
+  },
+  removed: {
+    beta: {
+      delta: {
+        sigma: 7,
+      },
+    },
+  },
+}
+*/
+```
+
+### flattenObject
+
+This utils returns a flattened version of the input object also accounting for nested properties.
+
+```ts
+const flattened = flattenObject({
+  alpha: {
+    gamma: {
+      sigma: 1,
+    },
+    delta: {
+      sigma: 2,
+    },
+  },
+  beta: 3,
+});
+
+/*
+Result:
+{
+  'alpha.gamma.sigma': 1,
+  'alpha.delta.sigma': 2,
+  beta: 3,
+}
+*/
+```
--- a/src/platform/packages/shared/kbn-object-utils/index.ts
+++ b/src/platform/packages/shared/kbn-object-utils/index.ts
@ -0,0 +1,11 @@
+/*
+ * Copyright Elasticsearch B.V. and/or licensed to Elasticsearch B.V. under one
+ * or more contributor license agreements. Licensed under the "Elastic License
+ * 2.0", the "GNU Affero General Public License v3.0 only", and the "Server Side
+ * Public License v 1"; you may not use this file except in compliance with, at
+ * your election, the "Elastic License 2.0", the "GNU Affero General Public
+ * License v3.0 only", or the "Server Side Public License, v 1".
+ */
+
+export * from './src/calculate_object_diff';
+export * from './src/flatten_object';
--- a/src/platform/packages/shared/kbn-object-utils/jest.config.js
+++ b/src/platform/packages/shared/kbn-object-utils/jest.config.js
@ -0,0 +1,14 @@
+/*
+ * Copyright Elasticsearch B.V. and/or licensed to Elasticsearch B.V. under one
+ * or more contributor license agreements. Licensed under the "Elastic License
+ * 2.0", the "GNU Affero General Public License v3.0 only", and the "Server Side
+ * Public License v 1"; you may not use this file except in compliance with, at
+ * your election, the "Elastic License 2.0", the "GNU Affero General Public
+ * License v3.0 only", or the "Server Side Public License, v 1".
+ */
+
+module.exports = {
+  preset: '@kbn/test',
+  rootDir: '../../../../..',
+  roots: ['<rootDir>/src/platform/packages/shared/kbn-object-utils'],
+};
--- a/src/platform/packages/shared/kbn-object-utils/kibana.jsonc
+++ b/src/platform/packages/shared/kbn-object-utils/kibana.jsonc
@ -0,0 +1,10 @@
+{
+  "type": "shared-common",
+  "id": "@kbn/object-utils",
+  "owner": [
+    "@elastic/kibana-core"
+  ],
+  "group": "platform",
+  "visibility": "shared",
+  "devOnly": false
+}
--- a/src/platform/packages/shared/kbn-object-utils/package.json
+++ b/src/platform/packages/shared/kbn-object-utils/package.json
@ -0,0 +1,8 @@
+{
+  "description": "Object utils for Kibana",
+  "license": "Elastic License 2.0 OR AGPL-3.0-only OR SSPL-1.0",
+  "name": "@kbn/object-utils",
+  "private": true,
+  "version": "1.0.0",
+  "sideEffects": false
+}
--- a/src/platform/packages/shared/kbn-object-utils/src/calculate_object_diff.test.ts
+++ b/src/platform/packages/shared/kbn-object-utils/src/calculate_object_diff.test.ts
@ -0,0 +1,34 @@
+/*
+ * Copyright Elasticsearch B.V. and/or licensed to Elasticsearch B.V. under one
+ * or more contributor license agreements. Licensed under the "Elastic License
+ * 2.0", the "GNU Affero General Public License v3.0 only", and the "Server Side
+ * Public License v 1"; you may not use this file except in compliance with, at
+ * your election, the "Elastic License 2.0", the "GNU Affero General Public
+ * License v3.0 only", or the "Server Side Public License, v 1".
+ */
+
+import { calculateObjectDiff } from './calculate_object_diff';
+
+describe('calculateObjectDiff', () => {
+  it('should return the added and removed parts between 2 objects', () => {
+    const { added, removed } = calculateObjectDiff({ alpha: 1, beta: 2 }, { alpha: 1, gamma: 3 });
+    expect(added).toEqual({ gamma: 3 });
+    expect(removed).toEqual({ beta: 2 });
+  });
+
+  it('should work on nested objects', () => {
+    const { added, removed } = calculateObjectDiff(
+      { alpha: 1, beta: { gamma: 2, delta: { sigma: 7 } } },
+      { alpha: 1, beta: { gamma: 2, eta: 4 } }
+    );
+
+    expect(added).toEqual({ beta: { eta: 4 } });
+    expect(removed).toEqual({ beta: { delta: { sigma: 7 } } });
+  });
+
+  it('should return empty added/removed when the objects are the same', () => {
+    const { added, removed } = calculateObjectDiff({ alpha: 1, beta: 2 }, { alpha: 1, beta: 2 });
+    expect(added).toEqual({});
+    expect(removed).toEqual({});
+  });
+});
--- a/src/platform/packages/shared/kbn-object-utils/src/calculate_object_diff.ts
+++ b/src/platform/packages/shared/kbn-object-utils/src/calculate_object_diff.ts
@ -0,0 +1,75 @@
+/*
+ * Copyright Elasticsearch B.V. and/or licensed to Elasticsearch B.V. under one
+ * or more contributor license agreements. Licensed under the "Elastic License
+ * 2.0", the "GNU Affero General Public License v3.0 only", and the "Server Side
+ * Public License v 1"; you may not use this file except in compliance with, at
+ * your election, the "Elastic License 2.0", the "GNU Affero General Public
+ * License v3.0 only", or the "Server Side Public License, v 1".
+ */
+
+import { isEmpty, isPlainObject } from 'lodash';
+
+interface Obj {
+  [key: PropertyKey]: Obj | unknown;
+}
+
+type DeepPartial<TInputObj> = {
+  [Prop in keyof TInputObj]?: TInputObj[Prop] extends Obj
+    ? DeepPartial<TInputObj[Prop]>
+    : TInputObj[Prop];
+};
+
+interface ObjectDiffResult<TBase, TCompare> {
+  added: DeepPartial<TCompare>;
+  removed: DeepPartial<TBase>;
+}
+
+/**
+ * Compares two JSON objects and calculates the added and removed properties, including nested properties.
+ * @param oldObj - The base object.
+ * @param newObj - The comparison object.
+ * @returns An object containing added and removed properties.
+ */
+export function calculateObjectDiff<TBase extends Obj, TCompare extends Obj>(
+  oldObj: TBase,
+  newObj?: TCompare
+): ObjectDiffResult<TBase, TCompare> {
+  const added: DeepPartial<TCompare> = {};
+  const removed: DeepPartial<TBase> = {};
+
+  if (!newObj) return { added, removed };
+
+  function diffRecursive(
+    base: Obj,
+    compare: Obj,
+    addedMap: DeepPartial<Obj>,
+    removedMap: DeepPartial<Obj>
+  ): void {
+    for (const key in compare) {
+      if (!(key in base)) {
+        addedMap[key] = compare[key];
+      } else if (isPlainObject(base[key]) && isPlainObject(compare[key])) {
+        addedMap[key] = {};
+        removedMap[key] = {};
+        diffRecursive(
+          base[key] as Obj,
+          compare[key] as Obj,
+          addedMap[key] as Obj,
+          removedMap[key] as Obj
+        );
+        if (isEmpty(addedMap[key])) delete addedMap[key];
+        if (isEmpty(removedMap[key])) delete removedMap[key];
+      }
+    }
+
+    for (const key in base) {
+      if (!(key in compare)) {
+        removedMap[key] = base[key];
+      }
+    }
+  }
+
+  diffRecursive(oldObj, newObj, added, removed);
+
+  return { added, removed };
+}
--- a/src/platform/packages/shared/kbn-object-utils/src/flatten_object.test.ts
+++ b/src/platform/packages/shared/kbn-object-utils/src/flatten_object.test.ts
@ -0,0 +1,32 @@
+/*
+ * Copyright Elasticsearch B.V. and/or licensed to Elasticsearch B.V. under one
+ * or more contributor license agreements. Licensed under the "Elastic License
+ * 2.0", the "GNU Affero General Public License v3.0 only", and the "Server Side
+ * Public License v 1"; you may not use this file except in compliance with, at
+ * your election, the "Elastic License 2.0", the "GNU Affero General Public
+ * License v3.0 only", or the "Server Side Public License, v 1".
+ */
+
+import { flattenObject } from './flatten_object';
+
+describe('flattenObject', () => {
+  it('should flat gamma object properties', () => {
+    const flattened = flattenObject({
+      alpha: {
+        gamma: {
+          sigma: 1,
+        },
+        delta: {
+          sigma: 2,
+        },
+      },
+      beta: 3,
+    });
+
+    expect(flattened).toEqual({
+      'alpha.gamma.sigma': 1,
+      'alpha.delta.sigma': 2,
+      beta: 3,
+    });
+  });
+});
--- a/src/platform/packages/shared/kbn-object-utils/src/flatten_object.ts
+++ b/src/platform/packages/shared/kbn-object-utils/src/flatten_object.ts
@ -0,0 +1,33 @@
+/*
+ * Copyright Elasticsearch B.V. and/or licensed to Elasticsearch B.V. under one
+ * or more contributor license agreements. Licensed under the "Elastic License
+ * 2.0", the "GNU Affero General Public License v3.0 only", and the "Server Side
+ * Public License v 1"; you may not use this file except in compliance with, at
+ * your election, the "Elastic License 2.0", the "GNU Affero General Public
+ * License v3.0 only", or the "Server Side Public License, v 1".
+ */
+
+import { isPlainObject } from 'lodash';
+
+/**
+ * Returns a flattened version of the input object also accounting for nested properties.
+ * @param obj - The input object.
+ * @param parentKey - The initial key used for recursive flattening.
+ * @returns An object containing all the flattened properties.
+ */
+export function flattenObject(obj: Record<PropertyKey, unknown>, parentKey: string = '') {
+  const result: Record<PropertyKey, unknown> = {};
+
+  for (const key in obj) {
+    if (Object.hasOwn(obj, key)) {
+      const value = obj[key];
+      const newKey = parentKey ? `${parentKey}.${key}` : key;
+      if (isPlainObject(value)) {
+        Object.assign(result, flattenObject(value as Record<PropertyKey, unknown>, newKey));
+      } else {
+        result[newKey] = value;
+      }
+    }
+  }
+  return result;
+}
--- a/src/platform/packages/shared/kbn-object-utils/tsconfig.json
+++ b/src/platform/packages/shared/kbn-object-utils/tsconfig.json
@ -0,0 +1,10 @@
+{
+  "compilerOptions": {
+    "outDir": "target/types",
+    "types": ["jest", "node"]
+  },
+  "exclude": ["target/**/*"],
+  "extends": "../../../../../tsconfig.base.json",
+  "include": ["**/*.ts"],
+  "kbn_references": []
+}
--- a/tsconfig.base.json
+++ b/tsconfig.base.json
@ -1312,6 +1312,8 @@
      "@kbn/no-data-page-plugin/*": ["src/platform/plugins/private/no_data_page/*"],
      "@kbn/notifications-plugin": ["x-pack/platform/plugins/shared/notifications"],
      "@kbn/notifications-plugin/*": ["x-pack/platform/plugins/shared/notifications/*"],
+      "@kbn/object-utils": ["src/platform/packages/shared/kbn-object-utils"],
+      "@kbn/object-utils/*": ["src/platform/packages/shared/kbn-object-utils/*"],
      "@kbn/object-versioning": ["src/platform/packages/shared/kbn-object-versioning"],
      "@kbn/object-versioning/*": ["src/platform/packages/shared/kbn-object-versioning/*"],
      "@kbn/object-versioning-utils": ["src/platform/packages/shared/kbn-object-versioning-utils"],
--- a/x-pack/packages/kbn-streams-schema/src/apis/list_streams_response.ts
+++ b/x-pack/packages/kbn-streams-schema/src/apis/list_streams_response.ts
@ -5,10 +5,10 @@
 * 2.0.
 */
 import { z } from '@kbn/zod';
-import { streamDefintionSchema } from '../models';
+import { streamDefinitionSchema } from '../models';

 export const listStreamsResponseSchema = z.object({
-  streams: z.array(streamDefintionSchema),
+  streams: z.array(streamDefinitionSchema),
 });

 export type ListStreamsResponse = z.infer<typeof listStreamsResponseSchema>;
--- a/x-pack/packages/kbn-streams-schema/src/helpers/index.ts
+++ b/x-pack/packages/kbn-streams-schema/src/helpers/index.ts
@ -5,4 +5,5 @@
 * 2.0.
 */

+export * from './processing';
 export * from './type_guards';
--- a/x-pack/packages/kbn-streams-schema/src/helpers/processing.ts
+++ b/x-pack/packages/kbn-streams-schema/src/helpers/processing.ts
@ -0,0 +1,38 @@
+/*
+ * Copyright Elasticsearch B.V. and/or licensed to Elasticsearch B.V. under one
+ * or more contributor license agreements. Licensed under the Elastic License
+ * 2.0; you may not use this file except in compliance with the Elastic License
+ * 2.0.
+ */
+
+import { Condition, ProcessingDefinition } from '../models';
+import {
+  isGrokProcessor,
+  isDissectProcessor,
+  isFilterCondition,
+  isAndCondition,
+  isOrCondition,
+} from './type_guards';
+
+export function getProcessorType(processor: ProcessingDefinition) {
+  if (isGrokProcessor(processor.config)) {
+    return 'grok';
+  }
+  if (isDissectProcessor(processor.config)) {
+    return 'dissect';
+  }
+  throw new Error('Unknown processor type');
+}
+
+export function isCompleteCondition(condition: Condition): boolean {
+  if (isFilterCondition(condition)) {
+    return condition.field !== undefined && condition.field !== '';
+  }
+  if (isAndCondition(condition)) {
+    return condition.and.every(isCompleteCondition);
+  }
+  if (isOrCondition(condition)) {
+    return condition.or.every(isCompleteCondition);
+  }
+  return false;
+}
--- a/x-pack/packages/kbn-streams-schema/src/helpers/type_guards.ts
+++ b/x-pack/packages/kbn-streams-schema/src/helpers/type_guards.ts
@ -10,7 +10,7 @@ import {
  AndCondition,
  conditionSchema,
  dissectProcessingDefinitionSchema,
-  DissectProcssingDefinition,
+  DissectProcessingDefinition,
  FilterCondition,
  filterConditionSchema,
  GrokProcessingDefinition,
@ -23,7 +23,7 @@ import {
  ReadStreamDefinition,
  readStreamDefinitonSchema,
  StreamDefinition,
-  streamDefintionSchema,
+  streamDefinitionSchema,
  WiredReadStreamDefinition,
  wiredReadStreamDefinitonSchema,
  WiredStreamDefinition,
@ -60,7 +60,7 @@ export function isIngestReadStream(subject: any): subject is IngestReadStreamDef
 }

 export function isStream(subject: any): subject is StreamDefinition {
-  return isSchema(streamDefintionSchema, subject);
+  return isSchema(streamDefinitionSchema, subject);
 }

 export function isIngestStream(subject: StreamDefinition): subject is IngestStreamDefinition {
@ -97,7 +97,7 @@ export function isGrokProcessor(subject: any): subject is GrokProcessingDefiniti
  return isSchema(grokProcessingDefinitionSchema, subject);
 }

-export function isDissectProcessor(subject: any): subject is DissectProcssingDefinition {
+export function isDissectProcessor(subject: any): subject is DissectProcessingDefinition {
  return isSchema(dissectProcessingDefinitionSchema, subject);
 }

--- a/x-pack/packages/kbn-streams-schema/src/models/common.ts
+++ b/x-pack/packages/kbn-streams-schema/src/models/common.ts
@ -52,6 +52,8 @@ export const grokProcessingDefinitionSchema = z.object({
    field: z.string(),
    patterns: z.array(z.string()),
    pattern_definitions: z.optional(z.record(z.string())),
+    ignore_failure: z.optional(z.boolean()),
+    ignore_missing: z.optional(z.boolean()),
  }),
 });

@ -61,10 +63,13 @@ export const dissectProcessingDefinitionSchema = z.object({
  dissect: z.object({
    field: z.string(),
    pattern: z.string(),
+    append_separator: z.optional(z.string()),
+    ignore_failure: z.optional(z.boolean()),
+    ignore_missing: z.optional(z.boolean()),
  }),
 });

-export type DissectProcssingDefinition = z.infer<typeof dissectProcessingDefinitionSchema>;
+export type DissectProcessingDefinition = z.infer<typeof dissectProcessingDefinitionSchema>;

 export const processingConfigSchema = z.union([
  grokProcessingDefinitionSchema,
@ -78,8 +83,24 @@ export const processingDefinitionSchema = z.object({

 export type ProcessingDefinition = z.infer<typeof processingDefinitionSchema>;

+export type ProcessorType = ProcessingDefinition['config'] extends infer U
+  ? U extends { [key: string]: any }
+    ? keyof U
+    : never
+  : never;
+
+export const FIELD_DEFINITION_TYPES = [
+  'keyword',
+  'match_only_text',
+  'long',
+  'double',
+  'date',
+  'boolean',
+  'ip',
+] as const;
+
 export const fieldDefinitionConfigSchema = z.object({
-  type: z.enum(['keyword', 'match_only_text', 'long', 'double', 'date', 'boolean', 'ip']),
+  type: z.enum(FIELD_DEFINITION_TYPES),
  format: z.optional(z.string()),
 });

--- a/x-pack/packages/kbn-streams-schema/src/models/streams/stream.ts
+++ b/x-pack/packages/kbn-streams-schema/src/models/streams/stream.ts
@ -9,9 +9,9 @@ import { z } from '@kbn/zod';
 import { wiredStreamDefinitonSchema } from './wired_stream';
 import { ingestStreamDefinitonSchema } from './ingest_stream';

-export const streamDefintionSchema = z.union([
+export const streamDefinitionSchema = z.union([
  wiredStreamDefinitonSchema,
  ingestStreamDefinitonSchema,
 ]);

-export type StreamDefinition = z.infer<typeof streamDefintionSchema>;
+export type StreamDefinition = z.infer<typeof streamDefinitionSchema>;
--- a/x-pack/solutions/observability/plugins/streams/server/lib/streams/errors/simulation_failed.ts
+++ b/x-pack/solutions/observability/plugins/streams/server/lib/streams/errors/simulation_failed.ts
@ -0,0 +1,20 @@
+/*
+ * Copyright Elasticsearch B.V. and/or licensed to Elasticsearch B.V. under one
+ * or more contributor license agreements. Licensed under the Elastic License
+ * 2.0; you may not use this file except in compliance with the Elastic License
+ * 2.0.
+ */
+
+import { errors } from '@elastic/elasticsearch';
+
+export class SimulationFailed extends Error {
+  constructor(error: errors.ResponseError) {
+    super(
+      error.body?.error?.reason ||
+        error.body?.error?.caused_by?.reason ||
+        error.message ||
+        'Unknown error'
+    );
+    this.name = 'SimulationFailed';
+  }
+}
--- a/x-pack/solutions/observability/plugins/streams/server/lib/streams/helpers/condition_fields.ts
+++ b/x-pack/solutions/observability/plugins/streams/server/lib/streams/helpers/condition_fields.ts
@ -13,19 +13,6 @@ import {
  isOrCondition,
 } from '@kbn/streams-schema';

-export function isComplete(condition: Condition): boolean {
-  if (isFilterCondition(condition)) {
-    return condition.field !== undefined && condition.field !== '';
-  }
-  if (isAndCondition(condition)) {
-    return condition.and.every(isComplete);
-  }
-  if (isOrCondition(condition)) {
-    return condition.or.every(isComplete);
-  }
-  return false;
-}
-
 export function getFields(
  condition: Condition
 ): Array<{ name: string; type: 'number' | 'string' }> {
--- a/x-pack/solutions/observability/plugins/streams/server/lib/streams/helpers/processing.ts
+++ b/x-pack/solutions/observability/plugins/streams/server/lib/streams/helpers/processing.ts
@ -0,0 +1,26 @@
+/*
+ * Copyright Elasticsearch B.V. and/or licensed to Elasticsearch B.V. under one
+ * or more contributor license agreements. Licensed under the Elastic License
+ * 2.0; you may not use this file except in compliance with the Elastic License
+ * 2.0.
+ */
+
+import { ProcessingDefinition, getProcessorType } from '@kbn/streams-schema';
+import { get } from 'lodash';
+import { IngestProcessorContainer } from '@elastic/elasticsearch/lib/api/types';
+import { conditionToPainless } from './condition_to_painless';
+
+export function formatToIngestProcessors(
+  processing: ProcessingDefinition[]
+): IngestProcessorContainer[] {
+  return processing.map((processor) => {
+    const type = getProcessorType(processor);
+    const config = get(processor.config, type);
+    return {
+      [type]: {
+        ...config,
+        if: processor.condition ? conditionToPainless(processor.condition) : undefined,
+      },
+    };
+  });
+}
--- a/x-pack/solutions/observability/plugins/streams/server/lib/streams/ingest_pipelines/generate_ingest_pipeline.ts
+++ b/x-pack/solutions/observability/plugins/streams/server/lib/streams/ingest_pipelines/generate_ingest_pipeline.ts
@ -5,48 +5,19 @@
 * 2.0.
 */

-import {
-  isDissectProcessor,
-  isGrokProcessor,
-  ProcessingDefinition,
-  StreamDefinition,
-} from '@kbn/streams-schema';
-import { get } from 'lodash';
+import { StreamDefinition } from '@kbn/streams-schema';
 import { ASSET_VERSION } from '../../../../common/constants';
-import { conditionToPainless } from '../helpers/condition_to_painless';
 import { logsDefaultPipelineProcessors } from './logs_default_pipeline';
 import { isRoot } from '../helpers/hierarchy';
 import { getProcessingPipelineName } from './name';
-
-function getProcessorType(processor: ProcessingDefinition) {
-  if (isGrokProcessor(processor.config)) {
-    return 'grok';
-  }
-  if (isDissectProcessor(processor.config)) {
-    return 'dissect';
-  }
-  throw new Error('Unknown processor type');
-}
-
-function generateProcessingSteps(definition: StreamDefinition) {
-  return definition.stream.ingest.processing.map((processor) => {
-    const type = getProcessorType(processor);
-    const config = get(processor.config, type);
-    return {
-      [type]: {
-        ...config,
-        if: processor.condition ? conditionToPainless(processor.condition) : undefined,
-      },
-    };
-  });
-}
+import { formatToIngestProcessors } from '../helpers/processing';

 export function generateIngestPipeline(id: string, definition: StreamDefinition) {
  return {
    id: getProcessingPipelineName(id),
    processors: [
      ...(isRoot(definition.name) ? logsDefaultPipelineProcessors : []),
-      ...generateProcessingSteps(definition),
+      ...formatToIngestProcessors(definition.stream.ingest.processing),
      {
        pipeline: {
          name: `${id}@stream.reroutes`,
@ -64,7 +35,7 @@ export function generateIngestPipeline(id: string, definition: StreamDefinition)

 export function generateClassicIngestPipelineBody(definition: StreamDefinition) {
  return {
-    processors: generateProcessingSteps(definition),
+    processors: formatToIngestProcessors(definition.stream.ingest.processing),
    _meta: {
      description: `Stream-managed pipeline for the ${definition.name} stream`,
      managed: true,
--- a/x-pack/solutions/observability/plugins/streams/server/routes/index.ts
+++ b/x-pack/solutions/observability/plugins/streams/server/routes/index.ts
@ -19,6 +19,7 @@ import { resyncStreamsRoute } from './streams/resync';
 import { sampleStreamRoute } from './streams/sample';
 import { schemaFieldsSimulationRoute } from './streams/schema/fields_simulation';
 import { unmappedFieldsRoute } from './streams/schema/unmapped_fields';
+import { simulateProcessorRoute } from './streams/processing/simulate';
 import { streamsStatusRoutes } from './streams/settings';

 export const streamsRouteRepository = {
@ -36,6 +37,7 @@ export const streamsRouteRepository = {
  ...sampleStreamRoute,
  ...streamDetailRoute,
  ...unmappedFieldsRoute,
+  ...simulateProcessorRoute,
  ...schemaFieldsSimulationRoute,
 };

--- a/x-pack/solutions/observability/plugins/streams/server/routes/streams/processing/simulate.ts
+++ b/x-pack/solutions/observability/plugins/streams/server/routes/streams/processing/simulate.ts
@ -0,0 +1,153 @@
+/*
+ * Copyright Elasticsearch B.V. and/or licensed to Elasticsearch B.V. under one
+ * or more contributor license agreements. Licensed under the Elastic License
+ * 2.0; you may not use this file except in compliance with the Elastic License
+ * 2.0.
+ */
+
+import { z } from '@kbn/zod';
+import { notFound, internal, badRequest } from '@hapi/boom';
+import { FieldDefinitionConfig, processingDefinitionSchema } from '@kbn/streams-schema';
+import { calculateObjectDiff, flattenObject } from '@kbn/object-utils';
+import {
+  IngestSimulateResponse,
+  IngestSimulateSimulateDocumentResult,
+} from '@elastic/elasticsearch/lib/api/types';
+import { SimulationFailed } from '../../../lib/streams/errors/simulation_failed';
+import { formatToIngestProcessors } from '../../../lib/streams/helpers/processing';
+import { createServerRoute } from '../../create_server_route';
+import { DefinitionNotFound } from '../../../lib/streams/errors';
+import { checkAccess } from '../../../lib/streams/stream_crud';
+
+export const simulateProcessorRoute = createServerRoute({
+  endpoint: 'POST /api/streams/{id}/processing/_simulate',
+  options: {
+    access: 'internal',
+  },
+  security: {
+    authz: {
+      enabled: false,
+      reason:
+        'This API delegates security to the currently logged in user and their Elasticsearch permissions.',
+    },
+  },
+  params: z.object({
+    path: z.object({ id: z.string() }),
+    body: z.object({
+      processing: z.array(processingDefinitionSchema),
+      documents: z.array(z.record(z.unknown())),
+    }),
+  }),
+  handler: async ({ params, request, response, getScopedClients }) => {
+    try {
+      const { scopedClusterClient } = await getScopedClients({ request });
+
+      const hasAccess = await checkAccess({ id: params.path.id, scopedClusterClient });
+      if (!hasAccess) {
+        throw new DefinitionNotFound(`Stream definition for ${params.path.id} not found.`);
+      }
+      // Normalize processing definition to pipeline processors
+      const processors = formatToIngestProcessors(params.body.processing);
+      // Convert input documents to ingest simulation format
+      const docs = params.body.documents.map((doc) => ({ _source: doc }));
+
+      let simulationResult: IngestSimulateResponse;
+      try {
+        simulationResult = await scopedClusterClient.asCurrentUser.ingest.simulate({
+          verbose: true,
+          pipeline: { processors },
+          docs,
+        });
+      } catch (error) {
+        throw new SimulationFailed(error);
+      }
+
+      const documents = computeSimulationDocuments(simulationResult, docs);
+      const detectedFields = computeDetectedFields(simulationResult, docs);
+      const successRate = computeSuccessRate(simulationResult);
+      const failureRate = 1 - successRate;
+
+      return {
+        documents,
+        success_rate: parseFloat(successRate.toFixed(2)),
+        failure_rate: parseFloat(failureRate.toFixed(2)),
+        detected_fields: detectedFields,
+      };
+    } catch (error) {
+      if (error instanceof DefinitionNotFound) {
+        throw notFound(error);
+      }
+
+      if (error instanceof SimulationFailed) {
+        throw badRequest(error);
+      }
+
+      throw internal(error);
+    }
+  },
+});
+
+const computeSimulationDocuments = (
+  simulation: IngestSimulateResponse,
+  sampleDocs: Array<{ _source: Record<string, unknown> }>
+) => {
+  return simulation.docs.map((doc, id) => {
+    // If every processor was successful, return and flatten the simulation doc from the last processor
+    if (isSuccessfulDocument(doc)) {
+      return {
+        value: flattenObject(doc.processor_results.at(-1)?.doc?._source ?? sampleDocs[id]._source),
+        isMatch: true,
+      };
+    }
+
+    return {
+      value: flattenObject(sampleDocs[id]._source),
+      isMatch: false,
+    };
+  });
+};
+
+const computeDetectedFields = (
+  simulation: IngestSimulateResponse,
+  sampleDocs: Array<{ _source: Record<string, unknown> }>
+): Array<{
+  name: string;
+  type: FieldDefinitionConfig['type'] | 'unmapped';
+}> => {
+  // Since we filter out failed documents, we need to map the simulation docs to the sample docs for later retrieval
+  const samplesToSimulationMap = new Map(simulation.docs.map((doc, id) => [doc, sampleDocs[id]]));
+
+  const diffs = simulation.docs
+    .filter(isSuccessfulDocument)
+    .map((doc) => {
+      const sample = samplesToSimulationMap.get(doc);
+      if (sample) {
+        const { added } = calculateObjectDiff(
+          sample._source,
+          doc.processor_results.at(-1)?.doc?._source
+        );
+        return flattenObject(added);
+      }
+
+      return {};
+    })
+    .map(Object.keys)
+    .flat();
+
+  const uniqueFields = [...new Set(diffs)];
+
+  return uniqueFields.map((name) => ({ name, type: 'unmapped' }));
+};
+
+const computeSuccessRate = (simulation: IngestSimulateResponse) => {
+  const successfulCount = simulation.docs.reduce((rate, doc) => {
+    return (rate += isSuccessfulDocument(doc) ? 1 : 0);
+  }, 0);
+  return successfulCount / simulation.docs.length;
+};
+
+const isSuccessfulDocument = (
+  doc: IngestSimulateSimulateDocumentResult
+): doc is Required<IngestSimulateSimulateDocumentResult> =>
+  doc.processor_results?.every((processorSimulation) => processorSimulation.status === 'success') ||
+  false;
--- a/x-pack/solutions/observability/plugins/streams/server/routes/streams/sample.ts
+++ b/x-pack/solutions/observability/plugins/streams/server/routes/streams/sample.ts
@ -8,12 +8,12 @@
 import { z } from '@kbn/zod';
 import { notFound, internal } from '@hapi/boom';
 import { errors } from '@elastic/elasticsearch';
-import { conditionSchema } from '@kbn/streams-schema';
+import { conditionSchema, isCompleteCondition } from '@kbn/streams-schema';
 import { createServerRoute } from '../create_server_route';
 import { DefinitionNotFound } from '../../lib/streams/errors';
 import { checkAccess } from '../../lib/streams/stream_crud';
 import { conditionToQueryDsl } from '../../lib/streams/helpers/condition_to_query_dsl';
-import { getFields, isComplete } from '../../lib/streams/helpers/condition_fields';
+import { getFields } from '../../lib/streams/helpers/condition_fields';

 export const sampleStreamRoute = createServerRoute({
  endpoint: 'POST /api/streams/{id}/_sample',
@ -48,7 +48,7 @@ export const sampleStreamRoute = createServerRoute({
        query: {
          bool: {
            must: [
-              isComplete(params.body.condition)
+              isCompleteCondition(params.body.condition)
                ? conditionToQueryDsl(params.body.condition)
                : { match_all: {} },
              {
--- a/x-pack/solutions/observability/plugins/streams/tsconfig.json
+++ b/x-pack/solutions/observability/plugins/streams/tsconfig.json
@ -28,6 +28,7 @@
    "@kbn/encrypted-saved-objects-plugin",
    "@kbn/licensing-plugin",
    "@kbn/server-route-repository-client",
+    "@kbn/object-utils",
    "@kbn/observability-utils-server",
    "@kbn/observability-utils-common",
    "@kbn/alerting-plugin",
--- a/x-pack/solutions/observability/plugins/streams_app/.storybook/get_mock_streams_app_context.tsx
+++ b/x-pack/solutions/observability/plugins/streams_app/.storybook/get_mock_streams_app_context.tsx
@ -17,9 +17,11 @@ import type { SavedObjectTaggingPluginStart } from '@kbn/saved-objects-tagging-p
 import type { StreamsAppKibanaContext } from '../public/hooks/use_kibana';

 export function getMockStreamsAppContext(): StreamsAppKibanaContext {
+  const appParams = coreMock.createAppMountParameters();
  const core = coreMock.createStart();

  return {
+    appParams,
    core,
    dependencies: {
      start: {
--- a/x-pack/solutions/observability/plugins/streams_app/public/components/app_root/index.tsx
+++ b/x-pack/solutions/observability/plugins/streams_app/public/components/app_root/index.tsx
@ -33,6 +33,7 @@ export function AppRoot({
  const { history } = appMountParameters;

  const context = {
+    appParams: appMountParameters,
    core: coreStart,
    dependencies: {
      start: pluginsStart,
--- a/x-pack/solutions/observability/plugins/streams_app/public/components/asset_image/index.tsx
+++ b/x-pack/solutions/observability/plugins/streams_app/public/components/asset_image/index.tsx
@ -0,0 +1,47 @@
+/*
+ * Copyright Elasticsearch B.V. and/or licensed to Elasticsearch B.V. under one
+ * or more contributor license agreements. Licensed under the Elastic License
+ * 2.0; you may not use this file except in compliance with the Elastic License
+ * 2.0.
+ */
+
+import { EuiImage, EuiImageProps, useEuiTheme } from '@elastic/eui';
+import { i18n } from '@kbn/i18n';
+import React, { useEffect } from 'react';
+import { useState } from 'react';
+
+const imageSets = {
+  welcome: {
+    light: import('./welcome_light.png'),
+    dark: import('./welcome_dark.png'),
+    alt: i18n.translate('xpack.streams.streamDetailView.welcomeImage', {
+      defaultMessage: 'Welcome image for the streams app',
+    }),
+  },
+  noResults: {
+    light: import('./no_results_light.png'),
+    dark: import('./no_results_dark.png'),
+    alt: i18n.translate('xpack.streams.streamDetailView.noResultsImage', {
+      defaultMessage: 'No results image for the streams app',
+    }),
+  },
+};
+
+interface AssetImageProps extends Omit<EuiImageProps, 'src' | 'url' | 'alt'> {
+  type?: keyof typeof imageSets;
+}
+
+export function AssetImage({ type = 'welcome', ...props }: AssetImageProps) {
+  const { colorMode } = useEuiTheme();
+  const { alt, dark, light } = imageSets[type];
+
+  const [imageSrc, setImageSrc] = useState<string>();
+
+  useEffect(() => {
+    const dynamicImageImport = colorMode === 'LIGHT' ? light : dark;
+
+    dynamicImageImport.then((module) => setImageSrc(module.default));
+  }, [colorMode, dark, light]);
+
+  return imageSrc ? <EuiImage size="l" {...props} alt={alt} src={imageSrc} /> : null;
+}
--- a/x-pack/solutions/observability/plugins/streams_app/public/components/asset_image/no_results_dark.png
+++ b/x-pack/solutions/observability/plugins/streams_app/public/components/asset_image/no_results_dark.png
--- a/x-pack/solutions/observability/plugins/streams_app/public/components/asset_image/no_results_light.png
+++ b/x-pack/solutions/observability/plugins/streams_app/public/components/asset_image/no_results_light.png
--- a/x-pack/solutions/observability/plugins/streams_app/public/components/asset_image/welcome_dark.png
+++ b/x-pack/solutions/observability/plugins/streams_app/public/components/asset_image/welcome_dark.png
--- a/x-pack/solutions/observability/plugins/streams_app/public/components/asset_image/welcome_light.png
+++ b/x-pack/solutions/observability/plugins/streams_app/public/components/asset_image/welcome_light.png
--- a/x-pack/solutions/observability/plugins/streams_app/public/components/assets/illustration.png
+++ b/x-pack/solutions/observability/plugins/streams_app/public/components/assets/illustration.png
--- a/x-pack/solutions/observability/plugins/streams_app/public/components/condition_editor/index.tsx
+++ b/x-pack/solutions/observability/plugins/streams_app/public/components/condition_editor/index.tsx
@ -175,11 +175,8 @@ function FilterForm(props: {
            };

            const newOperator = e.target.value as FilterCondition['operator'];
-            if (
-              'value' in newCondition &&
-              (newOperator === 'exists' || newOperator === 'notExists')
-            ) {
-              delete newCondition.value;
+            if (newOperator === 'exists' || newOperator === 'notExists') {
+              if ('value' in newCondition) delete newCondition.value;
            } else if (!('value' in newCondition)) {
              (newCondition as BinaryFilterCondition).value = '';
            }
--- a/x-pack/solutions/observability/plugins/streams_app/public/components/management_bottom_bar/index.tsx
+++ b/x-pack/solutions/observability/plugins/streams_app/public/components/management_bottom_bar/index.tsx
@ -0,0 +1,58 @@
+/*
+ * Copyright Elasticsearch B.V. and/or licensed to Elasticsearch B.V. under one
+ * or more contributor license agreements. Licensed under the Elastic License
+ * 2.0; you may not use this file except in compliance with the Elastic License
+ * 2.0.
+ */
+
+import React from 'react';
+import { EuiBottomBar, EuiButton, EuiButtonEmpty, EuiFlexGroup } from '@elastic/eui';
+import { i18n } from '@kbn/i18n';
+import { useDiscardConfirm } from '../../hooks/use_discard_confirm';
+
+interface ManagementBottomBarProps {
+  confirmButtonText?: string;
+  disabled?: boolean;
+  isLoading?: boolean;
+  onCancel: () => void;
+  onConfirm: () => void;
+}
+
+export function ManagementBottomBar({
+  confirmButtonText = defaultConfirmButtonText,
+  disabled = false,
+  isLoading = false,
+  onCancel,
+  onConfirm,
+}: ManagementBottomBarProps) {
+  const handleCancel = useDiscardConfirm(onCancel);
+
+  return (
+    <EuiBottomBar>
+      <EuiFlexGroup justifyContent="flexEnd" alignItems="center" responsive={false} gutterSize="s">
+        <EuiButtonEmpty color="text" size="s" iconType="cross" onClick={handleCancel}>
+          {i18n.translate('xpack.streams.streamDetailView.managementTab.bottomBar.cancel', {
+            defaultMessage: 'Cancel changes',
+          })}
+        </EuiButtonEmpty>
+
+        <EuiButton
+          disabled={disabled}
+          color="primary"
+          fill
+          size="s"
+          iconType="check"
+          onClick={onConfirm}
+          isLoading={isLoading}
+        >
+          {confirmButtonText}
+        </EuiButton>
+      </EuiFlexGroup>
+    </EuiBottomBar>
+  );
+}
+
+const defaultConfirmButtonText = i18n.translate(
+  'xpack.streams.streamDetailView.managementTab.bottomBar.confirm',
+  { defaultMessage: 'Save changes' }
+);
--- a/x-pack/solutions/observability/plugins/streams_app/public/components/preview_table/index.tsx
+++ b/x-pack/solutions/observability/plugins/streams_app/public/components/preview_table/index.tsx
@ -6,21 +6,23 @@
 */
 import { EuiDataGrid } from '@elastic/eui';
 import { i18n } from '@kbn/i18n';
-import React, { useEffect, useMemo, useState } from 'react';
+import React, { CSSProperties, useEffect, useMemo, useState } from 'react';

 export function PreviewTable({
  documents,
  displayColumns,
+  height,
 }: {
  documents: unknown[];
  displayColumns?: string[];
+  height?: CSSProperties['height'];
 }) {
-  const [height, setHeight] = useState('100px');
+  const [computedHeight, setComputedHeight] = useState('100px');
  useEffect(() => {
    // set height to 100% after a short delay otherwise it doesn't calculate correctly
    // TODO: figure out a better way to do this
    setTimeout(() => {
-      setHeight(`100%`);
+      setComputedHeight(`100%`);
    }, 50);
  }, []);

@ -59,7 +61,7 @@ export function PreviewTable({
      }}
      toolbarVisibility={false}
      rowCount={documents.length}
-      height={height}
+      height={height ?? computedHeight}
      renderCellValue={({ rowIndex, columnId }) => {
        const doc = documents[rowIndex];
        if (!doc || typeof doc !== 'object') {
--- a/x-pack/solutions/observability/plugins/streams_app/public/components/stream_detail_enriching/index.tsx
+++ b/x-pack/solutions/observability/plugins/streams_app/public/components/stream_detail_enriching/index.tsx
@ -1,18 +0,0 @@
-/*
- * Copyright Elasticsearch B.V. and/or licensed to Elasticsearch B.V. under one
- * or more contributor license agreements. Licensed under the Elastic License
- * 2.0; you may not use this file except in compliance with the Elastic License
- * 2.0.
- */
-import { StreamDefinition } from '@kbn/streams-schema';
-import React from 'react';
-
-export function StreamDetailEnriching({
-  definition: _definition,
-  refreshDefinition: _refreshDefinition,
-}: {
-  definition?: StreamDefinition;
-  refreshDefinition: () => void;
-}) {
-  return <>{'TODO'}</>;
-}
--- a/x-pack/solutions/observability/plugins/streams_app/public/components/stream_detail_enrichment/add_processor_button.tsx
+++ b/x-pack/solutions/observability/plugins/streams_app/public/components/stream_detail_enrichment/add_processor_button.tsx
@ -0,0 +1,22 @@
+/*
+ * Copyright Elasticsearch B.V. and/or licensed to Elasticsearch B.V. under one
+ * or more contributor license agreements. Licensed under the Elastic License
+ * 2.0; you may not use this file except in compliance with the Elastic License
+ * 2.0.
+ */
+
+import React from 'react';
+import { EuiButton } from '@elastic/eui';
+import { EuiButtonPropsForButton } from '@elastic/eui/src/components/button/button';
+import { i18n } from '@kbn/i18n';
+
+export function AddProcessorButton(props: EuiButtonPropsForButton) {
+  return (
+    <EuiButton iconType="plusInCircle" {...props}>
+      {i18n.translate(
+        'xpack.streams.streamDetailView.managementTab.enrichmentEmptyPrompt.addProcessorAction',
+        { defaultMessage: 'Add a processor' }
+      )}
+    </EuiButton>
+  );
+}
--- a/x-pack/solutions/observability/plugins/streams_app/public/components/stream_detail_enrichment/enrichment_empty_prompt.tsx
+++ b/x-pack/solutions/observability/plugins/streams_app/public/components/stream_detail_enrichment/enrichment_empty_prompt.tsx
@ -0,0 +1,44 @@
+/*
+ * Copyright Elasticsearch B.V. and/or licensed to Elasticsearch B.V. under one
+ * or more contributor license agreements. Licensed under the Elastic License
+ * 2.0; you may not use this file except in compliance with the Elastic License
+ * 2.0.
+ */
+
+import React from 'react';
+import { EuiEmptyPrompt } from '@elastic/eui';
+import { i18n } from '@kbn/i18n';
+import { AssetImage } from '../asset_image';
+import { AddProcessorButton } from './add_processor_button';
+
+interface EnrichmentEmptyPromptProps {
+  onAddProcessor: () => void;
+}
+
+export const EnrichmentEmptyPrompt = ({ onAddProcessor }: EnrichmentEmptyPromptProps) => {
+  return (
+    <EuiEmptyPrompt
+      titleSize="xs"
+      icon={<AssetImage />}
+      title={title}
+      body={body}
+      actions={[<AddProcessorButton onClick={onAddProcessor} />]}
+    />
+  );
+};
+
+const title = (
+  <h2>
+    {i18n.translate('xpack.streams.streamDetailView.managementTab.enrichmentEmptyPrompt.title', {
+      defaultMessage: 'Start extracting useful fields from your data',
+    })}
+  </h2>
+);
+
+const body = (
+  <p>
+    {i18n.translate('xpack.streams.streamDetailView.managementTab.enrichmentEmptyPrompt.body', {
+      defaultMessage: 'Use processors to transform data before indexing',
+    })}
+  </p>
+);
--- a/x-pack/solutions/observability/plugins/streams_app/public/components/stream_detail_enrichment/flyout/danger_zone.tsx
+++ b/x-pack/solutions/observability/plugins/streams_app/public/components/stream_detail_enrichment/flyout/danger_zone.tsx
@ -0,0 +1,89 @@
+/*
+ * Copyright Elasticsearch B.V. and/or licensed to Elasticsearch B.V. under one
+ * or more contributor license agreements. Licensed under the Elastic License
+ * 2.0; you may not use this file except in compliance with the Elastic License
+ * 2.0.
+ */
+
+import React from 'react';
+import {
+  EuiPanel,
+  EuiTitle,
+  EuiSpacer,
+  useGeneratedHtmlId,
+  EuiButton,
+  EuiConfirmModal,
+} from '@elastic/eui';
+import { i18n } from '@kbn/i18n';
+import { useBoolean } from '@kbn/react-hooks';
+
+export const DangerZone = ({
+  onDeleteProcessor,
+}: Pick<DeleteProcessorButtonProps, 'onDeleteProcessor'>) => {
+  return (
+    <EuiPanel hasShadow={false} paddingSize="none">
+      <EuiTitle size="xs">
+        <h3>
+          {i18n.translate(
+            'xpack.streams.streamDetailView.managementTab.enrichment.processorFlyout.dangerAreaTitle',
+            { defaultMessage: 'Danger area' }
+          )}
+        </h3>
+      </EuiTitle>
+      <EuiSpacer />
+      <DeleteProcessorButton onDeleteProcessor={onDeleteProcessor} />
+    </EuiPanel>
+  );
+};
+
+interface DeleteProcessorButtonProps {
+  onDeleteProcessor: () => void;
+}
+
+const DeleteProcessorButton = ({ onDeleteProcessor }: DeleteProcessorButtonProps) => {
+  const [isConfirmModalOpen, { on: openConfirmModal, off: closeConfirmModal }] = useBoolean();
+  const confirmModalId = useGeneratedHtmlId();
+
+  return (
+    <>
+      <EuiButton color="danger" onClick={openConfirmModal}>
+        {i18n.translate(
+          'xpack.streams.streamDetailView.managementTab.enrichment.processorFlyout.dangerAreaTitle',
+          { defaultMessage: 'Delete processor' }
+        )}
+      </EuiButton>
+      {isConfirmModalOpen && (
+        <EuiConfirmModal
+          aria-labelledby={confirmModalId}
+          title={i18n.translate(
+            'xpack.streams.streamDetailView.managementTab.enrichment.processorFlyout.deleteProcessorModalTitle',
+            { defaultMessage: 'Delete processor' }
+          )}
+          titleProps={{ id: confirmModalId }}
+          onCancel={closeConfirmModal}
+          onConfirm={onDeleteProcessor}
+          cancelButtonText={i18n.translate(
+            'xpack.streams.streamDetailView.managementTab.enrichment.processorFlyout.deleteProcessorModalCancel',
+            { defaultMessage: 'Keep processor' }
+          )}
+          confirmButtonText={i18n.translate(
+            'xpack.streams.streamDetailView.managementTab.enrichment.processorFlyout.deleteProcessorModalConfirm',
+            { defaultMessage: 'Delete processor' }
+          )}
+          buttonColor="danger"
+          defaultFocusedButton="confirm"
+        >
+          <p>
+            {i18n.translate(
+              'xpack.streams.streamDetailView.managementTab.enrichment.processorFlyout.deleteProcessorModalBody',
+              {
+                defaultMessage:
+                  'You can still reset this until the changes are confirmed on the processors list.',
+              }
+            )}
+          </p>
+        </EuiConfirmModal>
+      )}
+    </>
+  );
+};
--- a/x-pack/solutions/observability/plugins/streams_app/public/components/stream_detail_enrichment/flyout/dissect/dissect_append_separator.tsx
+++ b/x-pack/solutions/observability/plugins/streams_app/public/components/stream_detail_enrichment/flyout/dissect/dissect_append_separator.tsx
@ -0,0 +1,36 @@
+/*
+ * Copyright Elasticsearch B.V. and/or licensed to Elasticsearch B.V. under one
+ * or more contributor license agreements. Licensed under the Elastic License
+ * 2.0; you may not use this file except in compliance with the Elastic License
+ * 2.0.
+ */
+
+import React from 'react';
+import { useFormContext } from 'react-hook-form';
+import { EuiCode, EuiFieldText, EuiFormRow } from '@elastic/eui';
+import { i18n } from '@kbn/i18n';
+import { FormattedMessage } from '@kbn/i18n-react';
+
+export const DissectAppendSeparator = () => {
+  const { register } = useFormContext();
+  const { ref, ...inputProps } = register(`append_separator`);
+
+  return (
+    <EuiFormRow
+      label={i18n.translate(
+        'xpack.streams.streamDetailView.managementTab.enrichment.processorFlyout.dissectPatternSeparatorLabel',
+        { defaultMessage: 'Append separator' }
+      )}
+      helpText={
+        <FormattedMessage
+          id="xpack.streams.streamDetailView.managementTab.enrichment.processorFlyout.dissectPatternSeparatorHelpText"
+          defaultMessage="If you specify a key modifier, this character separates the fields when appending results. Defaults to {value}."
+          values={{ value: <EuiCode>&quot;&quot;</EuiCode> }}
+        />
+      }
+      fullWidth
+    >
+      <EuiFieldText {...inputProps} inputRef={ref} />
+    </EuiFormRow>
+  );
+};
--- a/x-pack/solutions/observability/plugins/streams_app/public/components/stream_detail_enrichment/flyout/dissect/dissect_pattern_definition.tsx
+++ b/x-pack/solutions/observability/plugins/streams_app/public/components/stream_detail_enrichment/flyout/dissect/dissect_pattern_definition.tsx
@ -0,0 +1,79 @@
+/*
+ * Copyright Elasticsearch B.V. and/or licensed to Elasticsearch B.V. under one
+ * or more contributor license agreements. Licensed under the Elastic License
+ * 2.0; you may not use this file except in compliance with the Elastic License
+ * 2.0.
+ */
+
+import React from 'react';
+import { useController } from 'react-hook-form';
+import { EuiFormRow, EuiLink } from '@elastic/eui';
+import { CodeEditor } from '@kbn/code-editor';
+import { i18n } from '@kbn/i18n';
+import { FormattedMessage } from '@kbn/i18n-react';
+import { useKibana } from '../../../../hooks/use_kibana';
+
+export const DissectPatternDefinition = () => {
+  const { core } = useKibana();
+  const esDocUrl = core.docLinks.links.ingest.dissectKeyModifiers;
+
+  const { field, fieldState } = useController({ name: 'pattern' });
+
+  return (
+    <EuiFormRow
+      label={i18n.translate(
+        'xpack.streams.streamDetailView.managementTab.enrichment.processorFlyout.dissectPatternDefinitionsLabel',
+        { defaultMessage: 'Pattern' }
+      )}
+      helpText={
+        <FormattedMessage
+          id="xpack.streams.streamDetailView.managementTab.enrichment.processorFlyout.dissectPatternDefinitionsHelpText"
+          defaultMessage="Pattern used to dissect the specified field. The pattern is defined by the parts of the string to discard. Use a {keyModifier} to alter the dissection behavior."
+          values={{
+            keyModifier: (
+              <EuiLink target="_blank" external href={esDocUrl}>
+                {i18n.translate(
+                  'xpack.streams.streamDetailView.managementTab.enrichment.processorFlyout.dissectPatternDefinitionsLink',
+                  { defaultMessage: 'key modifier' }
+                )}
+              </EuiLink>
+            ),
+          }}
+        />
+      }
+      isInvalid={fieldState.invalid}
+      fullWidth
+    >
+      <CodeEditor
+        value={serialize(field.value)}
+        onChange={(value) => field.onChange(deserialize(value))}
+        languageId="text"
+        height={75}
+        options={{ minimap: { enabled: false } }}
+        aria-label={i18n.translate(
+          'xpack.streams.streamDetailView.managementTab.enrichment.processorFlyout.dissectPatternDefinitionsAriaLabel',
+          { defaultMessage: 'Pattern editor' }
+        )}
+      />
+    </EuiFormRow>
+  );
+};
+
+const serialize = (input: string) => {
+  if (typeof input === 'string') {
+    const s = JSON.stringify(input);
+    return s.slice(1, s.length - 1);
+  }
+
+  return input;
+};
+
+const deserialize = (input: string) => {
+  if (typeof input === 'string') {
+    try {
+      return JSON.parse(`"${input}"`);
+    } catch (e) {
+      return input;
+    }
+  }
+};
--- a/x-pack/solutions/observability/plugins/streams_app/public/components/stream_detail_enrichment/flyout/dissect/index.tsx
+++ b/x-pack/solutions/observability/plugins/streams_app/public/components/stream_detail_enrichment/flyout/dissect/index.tsx
@ -0,0 +1,50 @@
+/*
+ * Copyright Elasticsearch B.V. and/or licensed to Elasticsearch B.V. under one
+ * or more contributor license agreements. Licensed under the Elastic License
+ * 2.0; you may not use this file except in compliance with the Elastic License
+ * 2.0.
+ */
+
+import React from 'react';
+import { EuiSpacer } from '@elastic/eui';
+import { i18n } from '@kbn/i18n';
+import { DissectAppendSeparator } from './dissect_append_separator';
+import { DissectPatternDefinition } from './dissect_pattern_definition';
+import { ProcessorFieldSelector } from '../processor_field_selector';
+import { ToggleField } from '../toggle_field';
+import { OptionalFieldsAccordion } from '../optional_fields_accordion';
+import { ProcessorConditionEditor } from '../processor_condition_editor';
+
+export const DissectProcessorForm = () => {
+  return (
+    <>
+      <ProcessorFieldSelector />
+      <DissectPatternDefinition />
+      <EuiSpacer size="m" />
+      <OptionalFieldsAccordion>
+        <DissectAppendSeparator />
+        <EuiSpacer size="m" />
+        <ProcessorConditionEditor />
+      </OptionalFieldsAccordion>
+      <EuiSpacer size="m" />
+      <ToggleField
+        name="ignore_failure"
+        label={i18n.translate(
+          'xpack.streams.streamDetailView.managementTab.enrichment.processorFlyout.ignoreFailuresLabel',
+          { defaultMessage: 'Ignore failures for this processor' }
+        )}
+      />
+      <ToggleField
+        name="ignore_missing"
+        label={i18n.translate(
+          'xpack.streams.streamDetailView.managementTab.enrichment.processorFlyout.ignoreMissingLabel',
+          { defaultMessage: 'Ignore missing' }
+        )}
+        helpText={i18n.translate(
+          'xpack.streams.streamDetailView.managementTab.enrichment.processorFlyout.ignoreMissingHelpText',
+          { defaultMessage: 'Ignore documents with a missing field.' }
+        )}
+      />
+    </>
+  );
+};
--- a/x-pack/solutions/observability/plugins/streams_app/public/components/stream_detail_enrichment/flyout/grok/grok_pattern_definition.tsx
+++ b/x-pack/solutions/observability/plugins/streams_app/public/components/stream_detail_enrichment/flyout/grok/grok_pattern_definition.tsx
@ -0,0 +1,92 @@
+/*
+ * Copyright Elasticsearch B.V. and/or licensed to Elasticsearch B.V. under one
+ * or more contributor license agreements. Licensed under the Elastic License
+ * 2.0; you may not use this file except in compliance with the Elastic License
+ * 2.0.
+ */
+
+import React from 'react';
+import { useController } from 'react-hook-form';
+import { EuiFormRow } from '@elastic/eui';
+import { CodeEditor } from '@kbn/code-editor';
+import { i18n } from '@kbn/i18n';
+
+export const GrokPatternDefinition = () => {
+  const { field, fieldState } = useController({ name: 'pattern_definitions' });
+
+  return (
+    <EuiFormRow
+      label={i18n.translate(
+        'xpack.streams.streamDetailView.managementTab.enrichment.processorFlyout.grokPatternDefinitionsLabel',
+        { defaultMessage: 'Pattern definitions' }
+      )}
+      helpText={i18n.translate(
+        'xpack.streams.streamDetailView.managementTab.enrichment.processorFlyout.grokPatternDefinitionsHelpText',
+        {
+          defaultMessage:
+            'A map of pattern-name and pattern tuples defining custom patterns. Patterns matching existing names will override the pre-existing definition.',
+        }
+      )}
+      isInvalid={fieldState.invalid}
+      fullWidth
+    >
+      <CodeEditor
+        value={serialize(field.value)}
+        onChange={(value) => field.onChange(deserialize(value))}
+        languageId="xjson"
+        height={200}
+        aria-label={i18n.translate(
+          'xpack.streams.streamDetailView.managementTab.enrichment.processorFlyout.grokPatternDefinitionsAriaLabel',
+          { defaultMessage: 'Pattern definitions editor' }
+        )}
+      />
+    </EuiFormRow>
+  );
+};
+
+const serialize = (v: unknown) => {
+  if (!v) {
+    return '{}';
+  }
+  if (typeof v === 'string') {
+    return formatXJsonString(v);
+  }
+  return JSON.stringify(v, null, 2);
+};
+
+const deserialize = (input: string) => {
+  try {
+    return JSON.parse(input);
+  } catch (e) {
+    return input;
+  }
+};
+
+/**
+ * Format a XJson string input as parsed JSON. Replaces the invalid characters
+ *  with a placeholder, parses the new string in a JSON format with the expected
+ * indentantion and then replaces the placeholders with the original values.
+ */
+const formatXJsonString = (input: string) => {
+  let placeholder = 'PLACEHOLDER';
+  const INVALID_STRING_REGEX = /"""(.*?)"""/gs;
+  while (input.includes(placeholder)) {
+    placeholder += '_';
+  }
+  const modifiedInput = input.replace(INVALID_STRING_REGEX, () => `"${placeholder}"`);
+
+  let jsonObject;
+  try {
+    jsonObject = JSON.parse(modifiedInput);
+  } catch (error) {
+    return input;
+  }
+  let formattedJsonString = JSON.stringify(jsonObject, null, 2);
+  const invalidStrings = input.match(INVALID_STRING_REGEX);
+  if (invalidStrings) {
+    invalidStrings.forEach((invalidString) => {
+      formattedJsonString = formattedJsonString.replace(`"${placeholder}"`, invalidString);
+    });
+  }
+  return formattedJsonString;
+};
--- a/x-pack/solutions/observability/plugins/streams_app/public/components/stream_detail_enrichment/flyout/grok/grok_patterns_editor.tsx
+++ b/x-pack/solutions/observability/plugins/streams_app/public/components/stream_detail_enrichment/flyout/grok/grok_patterns_editor.tsx
@ -0,0 +1,128 @@
+/*
+ * Copyright Elasticsearch B.V. and/or licensed to Elasticsearch B.V. under one
+ * or more contributor license agreements. Licensed under the Elastic License
+ * 2.0; you may not use this file except in compliance with the Elastic License
+ * 2.0.
+ */
+
+import React from 'react';
+import { useFormContext, useFieldArray, UseFormRegisterReturn } from 'react-hook-form';
+import {
+  DragDropContextProps,
+  EuiFormRow,
+  EuiPanel,
+  EuiSpacer,
+  EuiButtonEmpty,
+  EuiDraggable,
+  EuiFlexGroup,
+  EuiIcon,
+  EuiFieldText,
+  EuiButtonIcon,
+} from '@elastic/eui';
+import { i18n } from '@kbn/i18n';
+import { SortableList } from '../../sortable_list';
+import { GrokFormState } from '../../types';
+
+export const GrokPatternsEditor = () => {
+  const { register } = useFormContext();
+  const { fields, append, remove, move } = useFieldArray<Pick<GrokFormState, 'patterns'>>({
+    name: 'patterns',
+  });
+
+  const handlerPatternDrag: DragDropContextProps['onDragEnd'] = ({ source, destination }) => {
+    if (source && destination) {
+      move(source.index, destination.index);
+    }
+  };
+
+  const handleAddPattern = () => {
+    append({ value: '' });
+  };
+
+  const getRemovePatternHandler = (id: number) => (fields.length > 1 ? () => remove(id) : null);
+
+  return (
+    <EuiFormRow
+      label={i18n.translate(
+        'xpack.streams.streamDetailView.managementTab.enrichment.processorFlyout.grokEditorLabel',
+        { defaultMessage: 'Grok patterns editor' }
+      )}
+    >
+      <EuiPanel color="subdued" paddingSize="m">
+        <SortableList onDragItem={handlerPatternDrag}>
+          {fields.map((field, idx) => (
+            <DraggablePatternInput
+              key={field.id}
+              pattern={field}
+              idx={idx}
+              onRemove={getRemovePatternHandler(idx)}
+              inputProps={register(`patterns.${idx}.value`)}
+            />
+          ))}
+        </SortableList>
+        <EuiSpacer size="m" />
+        <EuiButtonEmpty onClick={handleAddPattern} iconType="plusInCircle" flush="left">
+          {i18n.translate(
+            'xpack.streams.streamDetailView.managementTab.enrichment.processorFlyout.grokEditor.addPattern',
+            { defaultMessage: 'Add pattern' }
+          )}
+        </EuiButtonEmpty>
+      </EuiPanel>
+    </EuiFormRow>
+  );
+};
+
+interface DraggablePatternInputProps {
+  idx: number;
+  inputProps: UseFormRegisterReturn<`patterns.${number}.value`>;
+  onRemove: ((idx: number) => void) | null;
+  pattern: GrokFormState['patterns'][number] & { id: string };
+}
+
+const DraggablePatternInput = ({
+  idx,
+  inputProps,
+  onRemove,
+  pattern,
+}: DraggablePatternInputProps) => {
+  const { ref, ...inputPropsWithoutRef } = inputProps;
+
+  return (
+    <EuiDraggable
+      index={idx}
+      spacing="m"
+      draggableId={pattern.id}
+      hasInteractiveChildren
+      customDragHandle
+      style={{
+        paddingLeft: 0,
+        paddingRight: 0,
+      }}
+    >
+      {(provided) => (
+        <EuiFlexGroup gutterSize="m" responsive={false} alignItems="center">
+          <EuiPanel
+            color="transparent"
+            paddingSize="xs"
+            {...provided.dragHandleProps}
+            aria-label="Drag Handle"
+          >
+            <EuiIcon type="grab" />
+          </EuiPanel>
+          <EuiFieldText {...inputPropsWithoutRef} inputRef={ref} compressed />
+          {onRemove && (
+            <EuiButtonIcon
+              iconType="minusInCircle"
+              color="danger"
+              onClick={() => onRemove(idx)}
+              aria-label={i18n.translate(
+                'xpack.streams.streamDetailView.managementTab.enrichment.processorFlyout.grokEditor.removePattern',
+                { defaultMessage: 'Remove grok pattern' }
+              )}
+            />
+          )}
+        </EuiFlexGroup>
+      )}
+    </EuiDraggable>
+  );
+};
--- a/x-pack/solutions/observability/plugins/streams_app/public/components/stream_detail_enrichment/flyout/grok/index.tsx
+++ b/x-pack/solutions/observability/plugins/streams_app/public/components/stream_detail_enrichment/flyout/grok/index.tsx
@ -0,0 +1,50 @@
+/*
+ * Copyright Elasticsearch B.V. and/or licensed to Elasticsearch B.V. under one
+ * or more contributor license agreements. Licensed under the Elastic License
+ * 2.0; you may not use this file except in compliance with the Elastic License
+ * 2.0.
+ */
+
+import React from 'react';
+import { EuiSpacer } from '@elastic/eui';
+import { i18n } from '@kbn/i18n';
+import { GrokPatternDefinition } from './grok_pattern_definition';
+import { GrokPatternsEditor } from './grok_patterns_editor';
+import { ProcessorFieldSelector } from '../processor_field_selector';
+import { ToggleField } from '../toggle_field';
+import { OptionalFieldsAccordion } from '../optional_fields_accordion';
+import { ProcessorConditionEditor } from '../processor_condition_editor';
+
+export const GrokProcessorForm = () => {
+  return (
+    <>
+      <ProcessorFieldSelector />
+      <GrokPatternsEditor />
+      <EuiSpacer size="m" />
+      <OptionalFieldsAccordion>
+        <GrokPatternDefinition />
+        <EuiSpacer size="m" />
+        <ProcessorConditionEditor />
+      </OptionalFieldsAccordion>
+      <EuiSpacer size="m" />
+      <ToggleField
+        name="ignore_failure"
+        label={i18n.translate(
+          'xpack.streams.streamDetailView.managementTab.enrichment.processorFlyout.ignoreFailuresLabel',
+          { defaultMessage: 'Ignore failures for this processor' }
+        )}
+      />
+      <ToggleField
+        name="ignore_missing"
+        label={i18n.translate(
+          'xpack.streams.streamDetailView.managementTab.enrichment.processorFlyout.ignoreMissingLabel',
+          { defaultMessage: 'Ignore missing' }
+        )}
+        helpText={i18n.translate(
+          'xpack.streams.streamDetailView.managementTab.enrichment.processorFlyout.ignoreMissingHelpText',
+          { defaultMessage: 'Ignore documents with a missing field.' }
+        )}
+      />
+    </>
+  );
+};
--- a/x-pack/solutions/observability/plugins/streams_app/public/components/stream_detail_enrichment/flyout/index.tsx
+++ b/x-pack/solutions/observability/plugins/streams_app/public/components/stream_detail_enrichment/flyout/index.tsx
@ -0,0 +1,169 @@
+/*
+ * Copyright Elasticsearch B.V. and/or licensed to Elasticsearch B.V. under one
+ * or more contributor license agreements. Licensed under the Elastic License
+ * 2.0; you may not use this file except in compliance with the Elastic License
+ * 2.0.
+ */
+
+import React, { useMemo } from 'react';
+import { FormProvider, SubmitHandler, useForm } from 'react-hook-form';
+import { EuiCallOut, EuiForm, EuiButton, EuiSpacer, EuiHorizontalRule } from '@elastic/eui';
+import { i18n } from '@kbn/i18n';
+import { ProcessingDefinition, ReadStreamDefinition, getProcessorType } from '@kbn/streams-schema';
+import { isEqual } from 'lodash';
+import { dynamic } from '@kbn/shared-ux-utility';
+import { ProcessorTypeSelector } from './processor_type_selector';
+import { ProcessorFlyoutTemplate } from './processor_flyout_template';
+import { DetectedField, ProcessorDefinition, ProcessorFormState } from '../types';
+import { DangerZone } from './danger_zone';
+import { DissectProcessorForm } from './dissect';
+import { GrokProcessorForm } from './grok';
+import { convertFormStateToProcessing, getDefaultFormState } from '../utils';
+
+const ProcessorOutcomePreview = dynamic(() =>
+  import(/* webpackChunkName: "management_processor_outcome" */ './processor_outcome_preview').then(
+    (mod) => ({
+      default: mod.ProcessorOutcomePreview,
+    })
+  )
+);
+
+export interface ProcessorFlyoutProps {
+  onClose: () => void;
+}
+
+export interface AddProcessorFlyoutProps extends ProcessorFlyoutProps {
+  definition: ReadStreamDefinition;
+  onAddProcessor: (newProcessing: ProcessingDefinition, newFields?: DetectedField[]) => void;
+}
+export interface EditProcessorFlyoutProps extends ProcessorFlyoutProps {
+  processor: ProcessorDefinition;
+  onDeleteProcessor: (id: string) => void;
+  onUpdateProcessor: (id: string, processor: ProcessorDefinition) => void;
+}
+
+export function AddProcessorFlyout({
+  definition,
+  onAddProcessor,
+  onClose,
+}: AddProcessorFlyoutProps) {
+  const defaultValues = useMemo(() => getDefaultFormState('grok'), []);
+
+  const methods = useForm<ProcessorFormState>({ defaultValues });
+
+  const formFields = methods.watch();
+
+  const hasChanges = useMemo(
+    () => !isEqual(defaultValues, formFields),
+    [defaultValues, formFields]
+  );
+
+  const handleSubmit: SubmitHandler<ProcessorFormState> = (data) => {
+    const processingDefinition = convertFormStateToProcessing(data);
+
+    onAddProcessor(processingDefinition, data.detected_fields);
+    onClose();
+  };
+
+  return (
+    <ProcessorFlyoutTemplate
+      shouldConfirm={hasChanges}
+      onClose={onClose}
+      title={i18n.translate(
+        'xpack.streams.streamDetailView.managementTab.enrichment.processorFlyout.titleAdd',
+        { defaultMessage: 'Add processor' }
+      )}
+      confirmButton={
+        <EuiButton onClick={methods.handleSubmit(handleSubmit)}>
+          {i18n.translate(
+            'xpack.streams.streamDetailView.managementTab.enrichment.processorFlyout.confirmAddProcessor',
+            { defaultMessage: 'Add processor' }
+          )}
+        </EuiButton>
+      }
+    >
+      <FormProvider {...methods}>
+        <EuiForm component="form" fullWidth onSubmit={methods.handleSubmit(handleSubmit)}>
+          <ProcessorTypeSelector />
+          <EuiSpacer size="m" />
+          {formFields.type === 'grok' && <GrokProcessorForm />}
+          {formFields.type === 'dissect' && <DissectProcessorForm />}
+        </EuiForm>
+        <EuiHorizontalRule />
+        <ProcessorOutcomePreview definition={definition} formFields={formFields} />
+      </FormProvider>
+    </ProcessorFlyoutTemplate>
+  );
+}
+
+export function EditProcessorFlyout({
+  onClose,
+  onDeleteProcessor,
+  onUpdateProcessor,
+  processor,
+}: EditProcessorFlyoutProps) {
+  const defaultValues = useMemo(
+    () => getDefaultFormState(getProcessorType(processor), processor),
+    [processor]
+  );
+
+  const methods = useForm<ProcessorFormState>({ defaultValues });
+
+  const formFields = methods.watch();
+
+  const hasChanges = useMemo(
+    () => !isEqual(defaultValues, formFields),
+    [defaultValues, formFields]
+  );
+
+  const handleSubmit: SubmitHandler<ProcessorFormState> = (data) => {
+    const processingDefinition = convertFormStateToProcessing(data);
+
+    onUpdateProcessor(processor.id, { id: processor.id, ...processingDefinition });
+    onClose();
+  };
+
+  const handleProcessorDelete = () => {
+    onDeleteProcessor(processor.id);
+    onClose();
+  };
+
+  return (
+    <ProcessorFlyoutTemplate
+      shouldConfirm={hasChanges}
+      onClose={onClose}
+      title={i18n.translate(
+        'xpack.streams.streamDetailView.managementTab.enrichment.processorFlyout.titleEdit',
+        { defaultMessage: 'Edit processor' }
+      )}
+      banner={
+        <EuiCallOut
+          title={i18n.translate(
+            'xpack.streams.streamDetailView.managementTab.enrichment.processorFlyout.calloutEdit',
+            { defaultMessage: 'Outcome preview is not available during edition' }
+          )}
+          iconType="iInCircle"
+        />
+      }
+      confirmButton={
+        <EuiButton onClick={methods.handleSubmit(handleSubmit)} disabled={!hasChanges}>
+          {i18n.translate(
+            'xpack.streams.streamDetailView.managementTab.enrichment.processorFlyout.confirmEditProcessor',
+            { defaultMessage: 'Update processor' }
+          )}
+        </EuiButton>
+      }
+    >
+      <FormProvider {...methods}>
+        <EuiForm component="form" fullWidth onSubmit={methods.handleSubmit(handleSubmit)}>
+          <ProcessorTypeSelector disabled />
+          <EuiSpacer size="m" />
+          {formFields.type === 'grok' && <GrokProcessorForm />}
+          {formFields.type === 'dissect' && <DissectProcessorForm />}
+          <EuiHorizontalRule />
+          <DangerZone onDeleteProcessor={handleProcessorDelete} />
+        </EuiForm>
+      </FormProvider>
+    </ProcessorFlyoutTemplate>
+  );
+}
--- a/x-pack/solutions/observability/plugins/streams_app/public/components/stream_detail_enrichment/flyout/optional_fields_accordion.tsx
+++ b/x-pack/solutions/observability/plugins/streams_app/public/components/stream_detail_enrichment/flyout/optional_fields_accordion.tsx
@ -0,0 +1,38 @@
+/*
+ * Copyright Elasticsearch B.V. and/or licensed to Elasticsearch B.V. under one
+ * or more contributor license agreements. Licensed under the Elastic License
+ * 2.0; you may not use this file except in compliance with the Elastic License
+ * 2.0.
+ */
+
+import React, { PropsWithChildren } from 'react';
+import { EuiAccordion, useEuiTheme } from '@elastic/eui';
+import { i18n } from '@kbn/i18n';
+import { css } from '@emotion/react';
+
+export const OptionalFieldsAccordion = ({ children }: PropsWithChildren) => {
+  const { euiTheme } = useEuiTheme();
+
+  return (
+    <EuiAccordion
+      element="fieldset"
+      id="optionalFieldsAccordion"
+      paddingSize="none"
+      buttonContent={i18n.translate(
+        'xpack.streams.streamDetailView.managementTab.enrichment.processorFlyout.optionalFields',
+        { defaultMessage: 'Optional fields' }
+      )}
+    >
+      <div
+        css={css`
+          border-left: ${euiTheme.border.thin};
+          margin-left: ${euiTheme.size.m};
+          padding-top: ${euiTheme.size.m};
+          padding-left: calc(${euiTheme.size.m} + ${euiTheme.size.xs});
+        `}
+      >
+        {children}
+      </div>
+    </EuiAccordion>
+  );
+};
--- a/x-pack/solutions/observability/plugins/streams_app/public/components/stream_detail_enrichment/flyout/processor_condition_editor.tsx
+++ b/x-pack/solutions/observability/plugins/streams_app/public/components/stream_detail_enrichment/flyout/processor_condition_editor.tsx
@ -0,0 +1,16 @@
+/*
+ * Copyright Elasticsearch B.V. and/or licensed to Elasticsearch B.V. under one
+ * or more contributor license agreements. Licensed under the Elastic License
+ * 2.0; you may not use this file except in compliance with the Elastic License
+ * 2.0.
+ */
+
+import React from 'react';
+import { useController } from 'react-hook-form';
+import { ConditionEditor } from '../../condition_editor';
+
+export const ProcessorConditionEditor = () => {
+  const { field } = useController({ name: 'condition' });
+
+  return <ConditionEditor condition={field.value} onConditionChange={field.onChange} />;
+};
--- a/x-pack/solutions/observability/plugins/streams_app/public/components/stream_detail_enrichment/flyout/processor_field_selector.tsx
+++ b/x-pack/solutions/observability/plugins/streams_app/public/components/stream_detail_enrichment/flyout/processor_field_selector.tsx
@ -0,0 +1,31 @@
+/*
+ * Copyright Elasticsearch B.V. and/or licensed to Elasticsearch B.V. under one
+ * or more contributor license agreements. Licensed under the Elastic License
+ * 2.0; you may not use this file except in compliance with the Elastic License
+ * 2.0.
+ */
+
+import { EuiFormRow, EuiFieldText } from '@elastic/eui';
+import { i18n } from '@kbn/i18n';
+import React from 'react';
+import { useFormContext } from 'react-hook-form';
+
+export const ProcessorFieldSelector = () => {
+  const { register } = useFormContext();
+  const { ref, ...inputProps } = register(`field`);
+
+  return (
+    <EuiFormRow
+      label={i18n.translate(
+        'xpack.streams.streamDetailView.managementTab.enrichment.processorFlyout.fieldSelectorLabel',
+        { defaultMessage: 'Field' }
+      )}
+      helpText={i18n.translate(
+        'xpack.streams.streamDetailView.managementTab.enrichment.processorFlyout.fieldSelectorHelpText',
+        { defaultMessage: 'Field to search for matches.' }
+      )}
+    >
+      <EuiFieldText {...inputProps} inputRef={ref} />
+    </EuiFormRow>
+  );
+};
--- a/x-pack/solutions/observability/plugins/streams_app/public/components/stream_detail_enrichment/flyout/processor_flyout_template.tsx
+++ b/x-pack/solutions/observability/plugins/streams_app/public/components/stream_detail_enrichment/flyout/processor_flyout_template.tsx
@ -0,0 +1,62 @@
+/*
+ * Copyright Elasticsearch B.V. and/or licensed to Elasticsearch B.V. under one
+ * or more contributor license agreements. Licensed under the Elastic License
+ * 2.0; you may not use this file except in compliance with the Elastic License
+ * 2.0.
+ */
+
+import React, { PropsWithChildren } from 'react';
+import {
+  EuiFlyoutResizable,
+  EuiFlyoutHeader,
+  EuiTitle,
+  EuiFlyoutBody,
+  EuiFlyoutFooter,
+  EuiFlexGroup,
+  EuiButtonEmpty,
+} from '@elastic/eui';
+import { i18n } from '@kbn/i18n';
+import { useDiscardConfirm } from '../../../hooks/use_discard_confirm';
+
+interface ProcessorFlyoutTemplateProps {
+  banner?: React.ReactNode;
+  confirmButton?: React.ReactNode;
+  onClose: () => void;
+  shouldConfirm?: boolean;
+  title: string;
+}
+
+export function ProcessorFlyoutTemplate({
+  banner,
+  children,
+  confirmButton,
+  onClose,
+  shouldConfirm = false,
+  title,
+}: PropsWithChildren<ProcessorFlyoutTemplateProps>) {
+  const handleClose = useDiscardConfirm(onClose);
+
+  const closeHandler = shouldConfirm ? handleClose : onClose;
+
+  return (
+    <EuiFlyoutResizable onClose={closeHandler}>
+      <EuiFlyoutHeader hasBorder>
+        <EuiTitle size="m">
+          <h2>{title}</h2>
+        </EuiTitle>
+      </EuiFlyoutHeader>
+      <EuiFlyoutBody banner={banner}>{children}</EuiFlyoutBody>
+      <EuiFlyoutFooter>
+        <EuiFlexGroup justifyContent="spaceBetween">
+          <EuiButtonEmpty iconType="cross" onClick={closeHandler}>
+            {i18n.translate(
+              'xpack.streams.streamDetailView.managementTab.enrichment.processorFlyout.cancel',
+              { defaultMessage: 'Cancel' }
+            )}
+          </EuiButtonEmpty>
+          {confirmButton}
+        </EuiFlexGroup>
+      </EuiFlyoutFooter>
+    </EuiFlyoutResizable>
+  );
+}
--- a/x-pack/solutions/observability/plugins/streams_app/public/components/stream_detail_enrichment/flyout/processor_outcome_preview.tsx
+++ b/x-pack/solutions/observability/plugins/streams_app/public/components/stream_detail_enrichment/flyout/processor_outcome_preview.tsx
@ -0,0 +1,459 @@
+/*
+ * Copyright Elasticsearch B.V. and/or licensed to Elasticsearch B.V. under one
+ * or more contributor license agreements. Licensed under the Elastic License
+ * 2.0; you may not use this file except in compliance with the Elastic License
+ * 2.0.
+ */
+
+import React, { useEffect, useMemo, useState } from 'react';
+import { useDateRange } from '@kbn/observability-utils-browser/hooks/use_date_range';
+import {
+  EuiPanel,
+  EuiTitle,
+  EuiSpacer,
+  EuiFlexGroup,
+  EuiFilterButton,
+  EuiFilterGroup,
+  EuiEmptyPrompt,
+  EuiLoadingLogo,
+  EuiButton,
+  EuiFormRow,
+  EuiSuperSelectOption,
+  EuiSuperSelect,
+  useEuiTheme,
+} from '@elastic/eui';
+import { i18n } from '@kbn/i18n';
+import { TimeRange } from '@kbn/es-query';
+import { isEmpty } from 'lodash';
+import { FieldIcon } from '@kbn/react-field';
+import {
+  FIELD_DEFINITION_TYPES,
+  ReadStreamDefinition,
+  isWiredReadStream,
+} from '@kbn/streams-schema';
+import { useController, useFieldArray } from 'react-hook-form';
+import { css } from '@emotion/react';
+import { flattenObject } from '@kbn/object-utils';
+import { IHttpFetchError, ResponseErrorBody } from '@kbn/core/public';
+import { useStreamsAppFetch } from '../../../hooks/use_streams_app_fetch';
+import { useKibana } from '../../../hooks/use_kibana';
+import { StreamsAppSearchBar, StreamsAppSearchBarProps } from '../../streams_app_search_bar';
+import { PreviewTable } from '../../preview_table';
+import { convertFormStateToProcessing, isCompleteProcessingDefinition } from '../utils';
+import { DetectedField, ProcessorFormState } from '../types';
+
+interface ProcessorOutcomePreviewProps {
+  definition: ReadStreamDefinition;
+  formFields: ProcessorFormState;
+}
+
+export const ProcessorOutcomePreview = ({
+  definition,
+  formFields,
+}: ProcessorOutcomePreviewProps) => {
+  const { dependencies } = useKibana();
+  const {
+    data,
+    streams: { streamsRepositoryClient },
+  } = dependencies.start;
+
+  const {
+    timeRange,
+    absoluteTimeRange: { start, end },
+    setTimeRange,
+  } = useDateRange({ data });
+
+  const [selectedDocsFilter, setSelectedDocsFilter] =
+    useState<DocsFilterOption>('outcome_filter_all');
+
+  const {
+    value: samples,
+    loading: isLoadingSamples,
+    refresh: refreshSamples,
+  } = useStreamsAppFetch(
+    ({ signal }) => {
+      if (!definition || !formFields.field) {
+        return { documents: [] };
+      }
+
+      return streamsRepositoryClient.fetch('POST /api/streams/{id}/_sample', {
+        signal,
+        params: {
+          path: { id: definition.name },
+          body: {
+            condition: { field: formFields.field, operator: 'exists' },
+            start: start?.valueOf(),
+            end: end?.valueOf(),
+            number: 100,
+          },
+        },
+      });
+    },
+    [definition, formFields.field, streamsRepositoryClient, start, end],
+    { disableToastOnError: true }
+  );
+
+  const {
+    value: simulation,
+    loading: isLoadingSimulation,
+    error,
+    refresh: refreshSimulation,
+  } = useStreamsAppFetch(
+    async ({ signal }) => {
+      if (!definition || !samples || isEmpty(samples.documents)) {
+        return Promise.resolve(null);
+      }
+
+      const processingDefinition = convertFormStateToProcessing(formFields);
+
+      if (!isCompleteProcessingDefinition(processingDefinition)) {
+        return Promise.resolve(null);
+      }
+
+      const simulationResult = await streamsRepositoryClient.fetch(
+        'POST /api/streams/{id}/processing/_simulate',
+        {
+          signal,
+          params: {
+            path: { id: definition.name },
+            body: {
+              documents: samples.documents as Array<Record<PropertyKey, unknown>>,
+              processing: [processingDefinition],
+            },
+          },
+        }
+      );
+
+      return simulationResult;
+    },
+    [definition, samples, streamsRepositoryClient],
+    { disableToastOnError: true }
+  );
+
+  const simulationError = error as IHttpFetchError<ResponseErrorBody> | undefined;
+
+  const simulationDocuments = useMemo(() => {
+    if (!simulation?.documents) {
+      const docs = (samples?.documents ?? []) as Array<Record<PropertyKey, unknown>>;
+      return docs.map((doc) => flattenObject(doc));
+    }
+
+    const filterDocuments = (filter: DocsFilterOption) => {
+      switch (filter) {
+        case 'outcome_filter_matched':
+          return simulation.documents.filter((doc) => doc.isMatch);
+        case 'outcome_filter_unmatched':
+          return simulation.documents.filter((doc) => !doc.isMatch);
+        case 'outcome_filter_all':
+        default:
+          return simulation.documents;
+      }
+    };
+
+    return filterDocuments(selectedDocsFilter).map((doc) => doc.value);
+  }, [samples?.documents, simulation?.documents, selectedDocsFilter]);
+
+  const detectedFieldsColumns = simulation?.detected_fields
+    ? simulation.detected_fields.map((field) => field.name)
+    : [];
+
+  const detectedFieldsEnabled =
+    isWiredReadStream(definition) && simulation && !isEmpty(simulation.detected_fields);
+
+  return (
+    <EuiPanel hasShadow={false} paddingSize="none">
+      <EuiFlexGroup alignItems="center" justifyContent="spaceBetween">
+        <EuiTitle size="xs">
+          <h3>
+            {i18n.translate(
+              'xpack.streams.streamDetailView.managementTab.enrichment.processorFlyout.outcomeTitle',
+              { defaultMessage: 'Outcome' }
+            )}
+          </h3>
+        </EuiTitle>
+        <EuiButton
+          iconType="play"
+          color="accentSecondary"
+          size="s"
+          onClick={refreshSimulation}
+          isLoading={isLoadingSimulation}
+        >
+          {i18n.translate(
+            'xpack.streams.streamDetailView.managementTab.enrichment.processorFlyout.runSimulation',
+            { defaultMessage: 'Run simulation' }
+          )}
+        </EuiButton>
+      </EuiFlexGroup>
+      <EuiSpacer />
+      {detectedFieldsEnabled && <DetectedFields detectedFields={simulation.detected_fields} />}
+      <OutcomeControls
+        docsFilter={selectedDocsFilter}
+        onDocsFilterChange={setSelectedDocsFilter}
+        timeRange={timeRange}
+        onTimeRangeChange={setTimeRange}
+        onTimeRangeRefresh={refreshSamples}
+        simulationFailureRate={simulation?.failure_rate}
+        simulationSuccessRate={simulation?.success_rate}
+      />
+      <EuiSpacer size="m" />
+      <OutcomePreviewTable
+        documents={simulationDocuments}
+        columns={[formFields.field, ...detectedFieldsColumns]}
+        error={simulationError}
+        isLoading={isLoadingSamples || isLoadingSimulation}
+      />
+    </EuiPanel>
+  );
+};
+
+const docsFilterOptions = {
+  outcome_filter_all: {
+    id: 'outcome_filter_all',
+    label: i18n.translate(
+      'xpack.streams.streamDetailView.managementTab.enrichment.processorFlyout.outcomeControls.all',
+      { defaultMessage: 'All samples' }
+    ),
+  },
+  outcome_filter_matched: {
+    id: 'outcome_filter_matched',
+    label: i18n.translate(
+      'xpack.streams.streamDetailView.managementTab.enrichment.processorFlyout.outcomeControls.matched',
+      { defaultMessage: 'Matched' }
+    ),
+  },
+  outcome_filter_unmatched: {
+    id: 'outcome_filter_unmatched',
+    label: i18n.translate(
+      'xpack.streams.streamDetailView.managementTab.enrichment.processorFlyout.outcomeControls.unmatched',
+      { defaultMessage: 'Unmatched' }
+    ),
+  },
+} as const;
+
+type DocsFilterOption = keyof typeof docsFilterOptions;
+
+interface OutcomeControlsProps {
+  docsFilter: DocsFilterOption;
+  timeRange: TimeRange;
+  onDocsFilterChange: (filter: DocsFilterOption) => void;
+  onTimeRangeChange: (timeRange: TimeRange) => void;
+  onTimeRangeRefresh: () => void;
+  simulationFailureRate?: number;
+  simulationSuccessRate?: number;
+}
+
+const OutcomeControls = ({
+  docsFilter,
+  timeRange,
+  onDocsFilterChange,
+  onTimeRangeChange,
+  onTimeRangeRefresh,
+  simulationFailureRate,
+  simulationSuccessRate,
+}: OutcomeControlsProps) => {
+  const handleQuerySubmit: StreamsAppSearchBarProps['onQuerySubmit'] = (
+    { dateRange },
+    isUpdate
+  ) => {
+    if (!isUpdate) {
+      return onTimeRangeRefresh();
+    }
+
+    if (dateRange) {
+      onTimeRangeChange({
+        from: dateRange.from,
+        to: dateRange?.to,
+        mode: dateRange.mode,
+      });
+    }
+  };
+
+  return (
+    <EuiFlexGroup alignItems="center" justifyContent="spaceBetween" wrap>
+      <EuiFilterGroup
+        aria-label={i18n.translate(
+          'xpack.streams.streamDetailView.managementTab.enrichment.processorFlyout.outcomeControlsAriaLabel',
+          { defaultMessage: 'Filter for all, matching or unmatching previewed documents.' }
+        )}
+      >
+        <EuiFilterButton
+          hasActiveFilters={docsFilter === docsFilterOptions.outcome_filter_all.id}
+          onClick={() => onDocsFilterChange(docsFilterOptions.outcome_filter_all.id)}
+        >
+          {docsFilterOptions.outcome_filter_all.label}
+        </EuiFilterButton>
+        <EuiFilterButton
+          hasActiveFilters={docsFilter === docsFilterOptions.outcome_filter_matched.id}
+          onClick={() => onDocsFilterChange(docsFilterOptions.outcome_filter_matched.id)}
+          badgeColor="success"
+          numActiveFilters={
+            simulationSuccessRate ? parseFloat((simulationSuccessRate * 100).toFixed(2)) : undefined
+          }
+        >
+          {docsFilterOptions.outcome_filter_matched.label}
+        </EuiFilterButton>
+        <EuiFilterButton
+          hasActiveFilters={docsFilter === docsFilterOptions.outcome_filter_unmatched.id}
+          onClick={() => onDocsFilterChange(docsFilterOptions.outcome_filter_unmatched.id)}
+          badgeColor="accent"
+          numActiveFilters={
+            simulationFailureRate ? parseFloat((simulationFailureRate * 100).toFixed(2)) : undefined
+          }
+        >
+          {docsFilterOptions.outcome_filter_unmatched.label}
+        </EuiFilterButton>
+      </EuiFilterGroup>
+      <StreamsAppSearchBar
+        onQuerySubmit={handleQuerySubmit}
+        onRefresh={onTimeRangeRefresh}
+        dateRangeFrom={timeRange.from}
+        dateRangeTo={timeRange.to}
+      />
+    </EuiFlexGroup>
+  );
+};
+
+const DetectedFields = ({ detectedFields }: { detectedFields: DetectedField[] }) => {
+  const { euiTheme } = useEuiTheme();
+  const { fields, replace } = useFieldArray<{ detected_fields: DetectedField[] }>({
+    name: 'detected_fields',
+  });
+
+  useEffect(() => {
+    replace(detectedFields);
+  }, [detectedFields, replace]);
+
+  return (
+    <EuiFormRow
+      label={i18n.translate(
+        'xpack.streams.streamDetailView.managementTab.enrichment.processorFlyout.detectedFieldsLabel',
+        { defaultMessage: 'Detected fields' }
+      )}
+      css={css`
+        margin-bottom: ${euiTheme.size.l};
+      `}
+      fullWidth
+    >
+      <EuiFlexGroup gutterSize="s" wrap>
+        {fields.map((field, id) => (
+          <DetectedFieldSelector key={field.name} selectorId={`detected_fields.${id}`} />
+        ))}
+      </EuiFlexGroup>
+    </EuiFormRow>
+  );
+};
+
+const DetectedFieldSelector = ({ selectorId }: { selectorId: string }) => {
+  const { field } = useController({ name: selectorId });
+
+  const options = useMemo(() => getDetectedFieldSelectOptions(field.value), [field]);
+
+  return (
+    <EuiSuperSelect
+      options={options}
+      valueOfSelected={field.value.type}
+      onChange={(type) => field.onChange({ ...field.value, type })}
+      css={css`
+        min-inline-size: 180px;
+      `}
+    />
+  );
+};
+
+const getDetectedFieldSelectOptions = (
+  fieldValue: DetectedField
+): Array<EuiSuperSelectOption<string>> =>
+  [...FIELD_DEFINITION_TYPES, 'unmapped'].map((type) => ({
+    value: type,
+    inputDisplay: (
+      <EuiFlexGroup alignItems="center" gutterSize="s">
+        <FieldIcon type={fieldValue.type} size="s" />
+        {fieldValue.name}
+      </EuiFlexGroup>
+    ),
+    dropdownDisplay: (
+      <EuiFlexGroup alignItems="center" gutterSize="s">
+        <FieldIcon type={type} size="s" />
+        {type}
+      </EuiFlexGroup>
+    ),
+  }));
+
+interface OutcomePreviewTableProps {
+  documents?: Array<Record<PropertyKey, unknown>>;
+  columns: string[];
+  error?: IHttpFetchError<ResponseErrorBody>;
+  isLoading?: boolean;
+}
+
+const OutcomePreviewTable = ({
+  documents = [],
+  columns,
+  error,
+  isLoading,
+}: OutcomePreviewTableProps) => {
+  if (error) {
+    return (
+      <EuiEmptyPrompt
+        iconType="error"
+        color="danger"
+        title={
+          <h3>
+            {i18n.translate(
+              'xpack.streams.streamDetailView.managementTab.enrichment.processorFlyout.outcomePreviewTable.errorTitle',
+              { defaultMessage: 'Unable to display the simulation outcome for this processor.' }
+            )}
+          </h3>
+        }
+        body={
+          <>
+            <p>
+              {i18n.translate(
+                'xpack.streams.streamDetailView.managementTab.enrichment.processorFlyout.outcomePreviewTable.errorBody',
+                { defaultMessage: 'The processor did not run correctly.' }
+              )}
+            </p>
+            {error.body?.message ? <p>{error.body.message}</p> : null}
+          </>
+        }
+      />
+    );
+  }
+
+  if (isLoading) {
+    return (
+      <EuiEmptyPrompt
+        icon={<EuiLoadingLogo logo="logoLogging" size="l" />}
+        title={
+          <h3>
+            {i18n.translate(
+              'xpack.streams.streamDetailView.managementTab.enrichment.processorFlyout.outcomePreviewTable.loadingTitle',
+              { defaultMessage: 'Running processor simulation' }
+            )}
+          </h3>
+        }
+      />
+    );
+  }
+
+  if (documents?.length === 0) {
+    return (
+      <EuiEmptyPrompt
+        iconType="dataVisualizer"
+        body={
+          <p>
+            {i18n.translate(
+              'xpack.streams.streamDetailView.managementTab.enrichment.processorFlyout.outcomePreviewTable.noDataTitle',
+              {
+                defaultMessage:
+                  'There are no simulation outcome documents for the current selection.',
+              }
+            )}
+          </p>
+        }
+      />
+    );
+  }
+
+  return <PreviewTable documents={documents} displayColumns={columns} height={500} />;
+};
--- a/x-pack/solutions/observability/plugins/streams_app/public/components/stream_detail_enrichment/flyout/processor_type_selector.tsx
+++ b/x-pack/solutions/observability/plugins/streams_app/public/components/stream_detail_enrichment/flyout/processor_type_selector.tsx
@ -0,0 +1,108 @@
+/*
+ * Copyright Elasticsearch B.V. and/or licensed to Elasticsearch B.V. under one
+ * or more contributor license agreements. Licensed under the Elastic License
+ * 2.0; you may not use this file except in compliance with the Elastic License
+ * 2.0.
+ */
+
+import React from 'react';
+import { EuiLink, EuiFormRow, EuiSuperSelect, EuiSuperSelectProps } from '@elastic/eui';
+import { i18n } from '@kbn/i18n';
+import { FormattedMessage } from '@kbn/i18n-react';
+import { useController, useFormContext, useWatch } from 'react-hook-form';
+import { ProcessorType } from '@kbn/streams-schema';
+import { useKibana } from '../../../hooks/use_kibana';
+import { getDefaultFormState } from '../utils';
+
+interface TAvailableProcessor {
+  value: ProcessorType;
+  inputDisplay: string;
+  getDocUrl: (esDocUrl: string) => React.ReactNode;
+}
+
+type TAvailableProcessors = Record<ProcessorType, TAvailableProcessor>;
+
+export const ProcessorTypeSelector = ({
+  disabled = false,
+}: Pick<EuiSuperSelectProps, 'disabled'>) => {
+  const { core } = useKibana();
+  const esDocUrl = core.docLinks.links.elasticsearch.docsBase;
+
+  const { control, reset } = useFormContext();
+  const { field, fieldState } = useController({ name: 'type', control, rules: { required: true } });
+
+  const processorType = useWatch<{ type: ProcessorType }>({ name: 'type' });
+
+  const handleChange = (type: ProcessorType) => {
+    const formState = getDefaultFormState(type);
+    reset(formState);
+  };
+
+  return (
+    <EuiFormRow
+      fullWidth
+      label={i18n.translate(
+        'xpack.streams.streamDetailView.managementTab.enrichment.processorFlyout.typeSelectorLabel',
+        { defaultMessage: 'Processor' }
+      )}
+      helpText={getProcessorDescription(esDocUrl)(processorType)}
+    >
+      <EuiSuperSelect
+        disabled={disabled}
+        options={processorTypeSelectorOptions}
+        isInvalid={fieldState.invalid}
+        valueOfSelected={field.value}
+        onChange={handleChange}
+        fullWidth
+        placeholder={i18n.translate(
+          'xpack.streams.streamDetailView.managementTab.enrichment.processorFlyout.typeSelectorPlaceholder',
+          { defaultMessage: 'Grok, Dissect ...' }
+        )}
+      />
+    </EuiFormRow>
+  );
+};
+
+const availableProcessors: TAvailableProcessors = {
+  dissect: {
+    value: 'dissect',
+    inputDisplay: 'Dissect',
+    getDocUrl: (esDocUrl: string) => (
+      <FormattedMessage
+        id="xpack.streams.streamDetailView.managementTab.enrichment.processorFlyout.dissectHelpText"
+        defaultMessage="Uses {dissectLink} patterns to extract matches from a field."
+        values={{
+          dissectLink: (
+            <EuiLink external target="_blank" href={esDocUrl + 'dissect-processor.html'}>
+              dissect
+            </EuiLink>
+          ),
+        }}
+      />
+    ),
+  },
+  grok: {
+    value: 'grok',
+    inputDisplay: 'Grok',
+    getDocUrl: (esDocUrl: string) => (
+      <FormattedMessage
+        id="xpack.streams.streamDetailView.managementTab.enrichment.processorFlyout.grokHelpText"
+        defaultMessage="Uses {grokLink} expressions to extract matches from a field."
+        values={{
+          grokLink: (
+            <EuiLink external target="_blank" href={esDocUrl + 'grok-processor.html'}>
+              grok
+            </EuiLink>
+          ),
+        }}
+      />
+    ),
+  },
+};
+
+const getProcessorDescription = (esDocUrl: string) => (type: ProcessorType) =>
+  availableProcessors[type].getDocUrl(esDocUrl);
+
+const processorTypeSelectorOptions = Object.values(availableProcessors).map(
+  ({ value, inputDisplay }) => ({ value, inputDisplay })
+);
--- a/x-pack/solutions/observability/plugins/streams_app/public/components/stream_detail_enrichment/flyout/toggle_field.tsx
+++ b/x-pack/solutions/observability/plugins/streams_app/public/components/stream_detail_enrichment/flyout/toggle_field.tsx
@ -0,0 +1,40 @@
+/*
+ * Copyright Elasticsearch B.V. and/or licensed to Elasticsearch B.V. under one
+ * or more contributor license agreements. Licensed under the Elastic License
+ * 2.0; you may not use this file except in compliance with the Elastic License
+ * 2.0.
+ */
+
+import React from 'react';
+import { useController } from 'react-hook-form';
+import { EuiFormRow, EuiSwitch, htmlIdGenerator } from '@elastic/eui';
+
+interface ToggleFieldProps {
+  helpText?: string;
+  id?: string;
+  label: string;
+  name: string;
+}
+
+export const ToggleField = ({
+  helpText,
+  id = createId(),
+  label,
+  name,
+  ...rest
+}: ToggleFieldProps) => {
+  const { field } = useController({ name });
+
+  return (
+    <EuiFormRow helpText={helpText} fullWidth describedByIds={id ? [id] : undefined} {...rest}>
+      <EuiSwitch
+        id={id}
+        label={label}
+        checked={field.value}
+        onChange={(e) => field.onChange(e.target.checked)}
+      />
+    </EuiFormRow>
+  );
+};
+
+const createId = htmlIdGenerator();
--- a/x-pack/solutions/observability/plugins/streams_app/public/components/stream_detail_enrichment/hooks/use_definition.ts
+++ b/x-pack/solutions/observability/plugins/streams_app/public/components/stream_detail_enrichment/hooks/use_definition.ts
@ -0,0 +1,162 @@
+/*
+ * Copyright Elasticsearch B.V. and/or licensed to Elasticsearch B.V. under one
+ * or more contributor license agreements. Licensed under the Elastic License
+ * 2.0; you may not use this file except in compliance with the Elastic License
+ * 2.0.
+ */
+
+import { useState, useMemo, useEffect } from 'react';
+import { i18n } from '@kbn/i18n';
+import { useAbortController } from '@kbn/observability-utils-browser/hooks/use_abort_controller';
+import { useBoolean } from '@kbn/react-hooks';
+import {
+  ReadStreamDefinition,
+  ProcessingDefinition,
+  isWiredReadStream,
+  FieldDefinition,
+  WiredReadStreamDefinition,
+} from '@kbn/streams-schema';
+import { htmlIdGenerator } from '@elastic/eui';
+import { isEqual } from 'lodash';
+import { DetectedField, ProcessorDefinition } from '../types';
+import { useKibana } from '../../../hooks/use_kibana';
+
+export const useDefinition = (definition: ReadStreamDefinition, refreshDefinition: () => void) => {
+  const { core, dependencies } = useKibana();
+
+  const { toasts } = core.notifications;
+  const { processing } = definition.stream.ingest;
+  const { streamsRepositoryClient } = dependencies.start.streams;
+
+  const abortController = useAbortController();
+  const [isSavingChanges, { on: startsSaving, off: endsSaving }] = useBoolean();
+
+  const [processors, setProcessors] = useState(() => createProcessorsList(processing));
+  const [fields, setFields] = useState(() =>
+    isWiredReadStream(definition) ? definition.stream.ingest.wired.fields : {}
+  );
+
+  const httpProcessing = useMemo(() => processors.map(removeIdFromProcessor), [processors]);
+
+  useEffect(() => {
+    // Reset processors when definition refreshes
+    setProcessors(createProcessorsList(definition.stream.ingest.processing));
+  }, [definition]);
+
+  const hasChanges = useMemo(
+    () => !isEqual(processing, httpProcessing),
+    [processing, httpProcessing]
+  );
+
+  const addProcessor = (newProcessing: ProcessingDefinition, newFields?: DetectedField[]) => {
+    setProcessors((prevProcs) => prevProcs.concat(createProcessorWithId(newProcessing)));
+
+    if (isWiredReadStream(definition) && newFields) {
+      setFields((currentFields) => mergeFields(definition, currentFields, newFields));
+    }
+  };
+
+  const updateProcessor = (id: string, processorUpdate: ProcessorDefinition) => {
+    setProcessors((prevProcs) =>
+      prevProcs.map((proc) => (proc.id === id ? processorUpdate : proc))
+    );
+  };
+
+  const deleteProcessor = (id: string) => {
+    setProcessors((prevProcs) => prevProcs.filter((proc) => proc.id !== id));
+  };
+
+  const resetChanges = () => {
+    setProcessors(createProcessorsList(processing));
+    setFields(isWiredReadStream(definition) ? definition.stream.ingest.wired.fields : {});
+  };
+
+  const saveChanges = async () => {
+    startsSaving();
+    try {
+      await streamsRepositoryClient.fetch(`PUT /api/streams/{id}`, {
+        signal: abortController.signal,
+        params: {
+          path: {
+            id: definition.name,
+          },
+          body: {
+            ingest: {
+              ...definition.stream.ingest,
+              processing: httpProcessing,
+              ...(isWiredReadStream(definition) && { wired: { fields } }),
+            },
+          },
+        },
+      });
+
+      toasts.addSuccess(
+        i18n.translate(
+          'xpack.streams.streamDetailView.managementTab.enrichment.saveChangesSuccess',
+          { defaultMessage: "Stream's processors updated" }
+        )
+      );
+    } catch (error) {
+      toasts.addError(error, {
+        title: i18n.translate(
+          'xpack.streams.streamDetailView.managementTab.enrichment.saveChangesError',
+          { defaultMessage: "An issue occurred saving processors' changes." }
+        ),
+        toastMessage: error.body.message,
+      });
+    } finally {
+      await refreshDefinition();
+      endsSaving();
+    }
+  };
+
+  return {
+    // Values
+    processors,
+    // Actions
+    addProcessor,
+    updateProcessor,
+    deleteProcessor,
+    resetChanges,
+    saveChanges,
+    setProcessors,
+    // Flags
+    hasChanges,
+    isSavingChanges,
+  };
+};
+
+const createId = htmlIdGenerator();
+const createProcessorsList = (processors: ProcessingDefinition[]): ProcessorDefinition[] =>
+  processors.map(createProcessorWithId);
+
+const createProcessorWithId = (processor: ProcessingDefinition): ProcessorDefinition => ({
+  ...processor,
+  id: createId(),
+});
+
+const removeIdFromProcessor = (processor: ProcessorDefinition): ProcessingDefinition => {
+  const { id, ...rest } = processor;
+  return rest;
+};
+
+const mergeFields = (
+  definition: WiredReadStreamDefinition,
+  currentFields: FieldDefinition,
+  newFields: DetectedField[]
+) => {
+  return {
+    ...definition.stream.ingest.wired.fields,
+    ...newFields.reduce((acc, field) => {
+      // Add only new fields and ignore unmapped ones
+      if (
+        !(field.name in currentFields) &&
+        !(field.name in definition.inherited_fields) &&
+        field.type !== 'unmapped'
+      ) {
+        acc[field.name] = { type: field.type };
+      }
+      return acc;
+    }, {} as FieldDefinition),
+  };
+};
--- a/x-pack/solutions/observability/plugins/streams_app/public/components/stream_detail_enrichment/index.tsx
+++ b/x-pack/solutions/observability/plugins/streams_app/public/components/stream_detail_enrichment/index.tsx
@ -0,0 +1,31 @@
+/*
+ * Copyright Elasticsearch B.V. and/or licensed to Elasticsearch B.V. under one
+ * or more contributor license agreements. Licensed under the Elastic License
+ * 2.0; you may not use this file except in compliance with the Elastic License
+ * 2.0.
+ */
+import React from 'react';
+import { dynamic } from '@kbn/shared-ux-utility';
+import { ReadStreamDefinition } from '@kbn/streams-schema';
+
+const StreamDetailEnrichmentContent = dynamic(() =>
+  import(/* webpackChunkName: "management_enrichment" */ './page_content').then((mod) => ({
+    default: mod.StreamDetailEnrichmentContent,
+  }))
+);
+
+interface StreamDetailEnrichmentProps {
+  definition?: ReadStreamDefinition;
+  refreshDefinition: () => void;
+}
+
+export function StreamDetailEnrichment({
+  definition,
+  refreshDefinition,
+}: StreamDetailEnrichmentProps) {
+  if (!definition) return null;
+
+  return (
+    <StreamDetailEnrichmentContent definition={definition} refreshDefinition={refreshDefinition} />
+  );
+}
--- a/x-pack/solutions/observability/plugins/streams_app/public/components/stream_detail_enrichment/page_content.tsx
+++ b/x-pack/solutions/observability/plugins/streams_app/public/components/stream_detail_enrichment/page_content.tsx
@ -0,0 +1,157 @@
+/*
+ * Copyright Elasticsearch B.V. and/or licensed to Elasticsearch B.V. under one
+ * or more contributor license agreements. Licensed under the Elastic License
+ * 2.0; you may not use this file except in compliance with the Elastic License
+ * 2.0.
+ */
+
+import React, { useEffect } from 'react';
+import {
+  DragDropContextProps,
+  EuiPanel,
+  EuiSpacer,
+  EuiText,
+  EuiTitle,
+  euiDragDropReorder,
+} from '@elastic/eui';
+import { i18n } from '@kbn/i18n';
+import { ReadStreamDefinition, isRootStream } from '@kbn/streams-schema';
+import { useBoolean } from '@kbn/react-hooks';
+import { useUnsavedChangesPrompt } from '@kbn/unsaved-changes-prompt';
+import { EnrichmentEmptyPrompt } from './enrichment_empty_prompt';
+import { AddProcessorButton } from './add_processor_button';
+import { AddProcessorFlyout } from './flyout';
+import { DraggableProcessorListItem } from './processors_list';
+import { ManagementBottomBar } from '../management_bottom_bar';
+import { SortableList } from './sortable_list';
+import { useDefinition } from './hooks/use_definition';
+import { useKibana } from '../../hooks/use_kibana';
+import { RootStreamEmptyPrompt } from './root_stream_empty_prompt';
+
+interface StreamDetailEnrichmentContentProps {
+  definition: ReadStreamDefinition;
+  refreshDefinition: () => void;
+}
+
+export function StreamDetailEnrichmentContent({
+  definition,
+  refreshDefinition,
+}: StreamDetailEnrichmentContentProps) {
+  const { appParams, core } = useKibana();
+
+  const [isBottomBarOpen, { on: openBottomBar, off: closeBottomBar }] = useBoolean();
+  const [isAddProcessorOpen, { on: openAddProcessor, off: closeAddProcessor }] = useBoolean();
+
+  const {
+    processors,
+    addProcessor,
+    updateProcessor,
+    deleteProcessor,
+    resetChanges,
+    saveChanges,
+    setProcessors,
+    hasChanges,
+    isSavingChanges,
+  } = useDefinition(definition, refreshDefinition);
+
+  const handlerItemDrag: DragDropContextProps['onDragEnd'] = ({ source, destination }) => {
+    if (source && destination) {
+      const items = euiDragDropReorder(processors, source.index, destination.index);
+      setProcessors(items);
+    }
+  };
+
+  useEffect(() => {
+    if (hasChanges) openBottomBar();
+    else closeBottomBar();
+  }, [closeBottomBar, hasChanges, openBottomBar]);
+
+  useUnsavedChangesPrompt({
+    hasUnsavedChanges: hasChanges,
+    history: appParams.history,
+    http: core.http,
+    navigateToUrl: core.application.navigateToUrl,
+    openConfirm: core.overlays.openConfirm,
+  });
+
+  const handleSaveChanges = async () => {
+    await saveChanges();
+    closeBottomBar();
+  };
+
+  const handleDiscardChanges = async () => {
+    await resetChanges();
+    closeBottomBar();
+  };
+
+  const bottomBar = isBottomBarOpen && (
+    <ManagementBottomBar
+      onCancel={handleDiscardChanges}
+      onConfirm={handleSaveChanges}
+      isLoading={isSavingChanges}
+    />
+  );
+
+  const addProcessorFlyout = isAddProcessorOpen && (
+    <AddProcessorFlyout
+      key="add-processor"
+      definition={definition}
+      onClose={closeAddProcessor}
+      onAddProcessor={addProcessor}
+    />
+  );
+
+  const hasProcessors = processors.length > 0;
+
+  if (isRootStream(definition)) {
+    return <RootStreamEmptyPrompt />;
+  }
+
+  return (
+    <>
+      {hasProcessors ? (
+        <EuiPanel paddingSize="none">
+          <ProcessorsHeader />
+          <EuiSpacer size="l" />
+          <SortableList onDragItem={handlerItemDrag}>
+            {processors.map((processor, idx) => (
+              <DraggableProcessorListItem
+                key={processor.id}
+                idx={idx}
+                processor={processor}
+                onUpdateProcessor={updateProcessor}
+                onDeleteProcessor={deleteProcessor}
+              />
+            ))}
+          </SortableList>
+          <EuiSpacer size="m" />
+          <AddProcessorButton onClick={openAddProcessor} />
+        </EuiPanel>
+      ) : (
+        <EnrichmentEmptyPrompt onAddProcessor={openAddProcessor} />
+      )}
+      {addProcessorFlyout}
+      {bottomBar}
+    </>
+  );
+}
+
+const ProcessorsHeader = () => {
+  return (
+    <>
+      <EuiTitle size="xs">
+        <h2>
+          {i18n.translate('xpack.streams.streamDetailView.managementTab.enrichment.headingTitle', {
+            defaultMessage: 'Processors for field extraction',
+          })}
+        </h2>
+      </EuiTitle>
+      <EuiText component="p" size="s">
+        {i18n.translate('xpack.streams.streamDetailView.managementTab.enrichment.headingSubtitle', {
+          defaultMessage:
+            'Use processors to transform data before indexing. Drag and drop existing processors to update their execution order.',
+        })}
+      </EuiText>
+    </>
+  );
+};
--- a/x-pack/solutions/observability/plugins/streams_app/public/components/stream_detail_enrichment/processors_list.tsx
+++ b/x-pack/solutions/observability/plugins/streams_app/public/components/stream_detail_enrichment/processors_list.tsx
@ -0,0 +1,114 @@
+/*
+ * Copyright Elasticsearch B.V. and/or licensed to Elasticsearch B.V. under one
+ * or more contributor license agreements. Licensed under the Elastic License
+ * 2.0; you may not use this file except in compliance with the Elastic License
+ * 2.0.
+ */
+
+import React from 'react';
+import {
+  EuiDraggable,
+  EuiPanelProps,
+  EuiPanel,
+  EuiFlexGroup,
+  EuiIcon,
+  EuiText,
+  EuiFlexItem,
+  EuiButtonIcon,
+} from '@elastic/eui';
+import { i18n } from '@kbn/i18n';
+import { getProcessorType, isDissectProcessor, isGrokProcessor } from '@kbn/streams-schema';
+import { useBoolean } from '@kbn/react-hooks';
+import { css } from '@emotion/react';
+import { EditProcessorFlyout, EditProcessorFlyoutProps } from './flyout';
+import { ProcessorDefinition } from './types';
+
+export const DraggableProcessorListItem = ({
+  processor,
+  idx,
+  ...props
+}: Omit<ProcessorListItemProps, 'hasShadow'> & { idx: number }) => (
+  <EuiDraggable
+    index={idx}
+    spacing="m"
+    draggableId={processor.id}
+    hasInteractiveChildren
+    style={{
+      paddingLeft: 0,
+      paddingRight: 0,
+    }}
+  >
+    {(_provided, state) => (
+      <ProcessorListItem processor={processor} hasShadow={state.isDragging} {...props} />
+    )}
+  </EuiDraggable>
+);
+
+interface ProcessorListItemProps {
+  processor: ProcessorDefinition;
+  hasShadow: EuiPanelProps['hasShadow'];
+  onUpdateProcessor: EditProcessorFlyoutProps['onUpdateProcessor'];
+  onDeleteProcessor: EditProcessorFlyoutProps['onDeleteProcessor'];
+}
+
+const ProcessorListItem = ({
+  processor,
+  hasShadow = false,
+  onUpdateProcessor,
+  onDeleteProcessor,
+}: ProcessorListItemProps) => {
+  const [isEditProcessorOpen, { on: openEditProcessor, off: closeEditProcessor }] = useBoolean();
+
+  const type = getProcessorType(processor);
+  const description = getProcessorDescription(processor);
+
+  return (
+    <EuiPanel hasBorder hasShadow={hasShadow} paddingSize="s">
+      <EuiFlexGroup gutterSize="m" responsive={false} alignItems="center">
+        <EuiIcon type="grab" />
+        <EuiText component="span" size="s">
+          {type.toUpperCase()}
+        </EuiText>
+        <EuiFlexItem
+          /* Allow text to overflow in flex child nodes */
+          css={css`
+            min-width: 0;
+          `}
+        >
+          <EuiText component="span" size="s" color="subdued" className="eui-textTruncate">
+            {description}
+          </EuiText>
+        </EuiFlexItem>
+        <EuiButtonIcon
+          onClick={openEditProcessor}
+          iconType="pencil"
+          color="text"
+          size="s"
+          aria-label={i18n.translate(
+            'xpack.streams.streamDetailView.managementTab.enrichment.editProcessorAction',
+            { defaultMessage: 'Edit {type} processor', values: { type } }
+          )}
+        />
+      </EuiFlexGroup>
+      {isEditProcessorOpen && (
+        <EditProcessorFlyout
+          key={`edit-processor`}
+          processor={processor}
+          onClose={closeEditProcessor}
+          onUpdateProcessor={onUpdateProcessor}
+          onDeleteProcessor={onDeleteProcessor}
+        />
+      )}
+    </EuiPanel>
+  );
+};
+
+const getProcessorDescription = (processor: ProcessorDefinition) => {
+  if (isGrokProcessor(processor.config)) {
+    return processor.config.grok.patterns.join(' • ');
+  } else if (isDissectProcessor(processor.config)) {
+    return processor.config.dissect.pattern;
+  }
+
+  return '';
+};
--- a/x-pack/solutions/observability/plugins/streams_app/public/components/stream_detail_enrichment/root_stream_empty_prompt.tsx
+++ b/x-pack/solutions/observability/plugins/streams_app/public/components/stream_detail_enrichment/root_stream_empty_prompt.tsx
@ -0,0 +1,39 @@
+/*
+ * Copyright Elasticsearch B.V. and/or licensed to Elasticsearch B.V. under one
+ * or more contributor license agreements. Licensed under the Elastic License
+ * 2.0; you may not use this file except in compliance with the Elastic License
+ * 2.0.
+ */
+
+import React from 'react';
+import { EuiEmptyPrompt } from '@elastic/eui';
+import { i18n } from '@kbn/i18n';
+import { AssetImage } from '../asset_image';
+
+export const RootStreamEmptyPrompt = () => {
+  return (
+    <EuiEmptyPrompt
+      titleSize="xs"
+      icon={<AssetImage type="noResults" />}
+      title={
+        <h2>
+          {i18n.translate(
+            'xpack.streams.streamDetailView.managementTab.rootStreamEmptyPrompt.title',
+            { defaultMessage: 'Processing data is not allowed for root streams.' }
+          )}
+        </h2>
+      }
+      body={
+        <p>
+          {i18n.translate(
+            'xpack.streams.streamDetailView.managementTab.rootStreamEmptyPrompt.body',
+            {
+              defaultMessage:
+                'Root streams are selectively immutable and cannot be enriched with processors. To enrich data, reroute a new child stream and add processors to it.',
+            }
+          )}
+        </p>
+      }
+    />
+  );
+};
--- a/x-pack/solutions/observability/plugins/streams_app/public/components/stream_detail_enrichment/sortable_list.tsx
+++ b/x-pack/solutions/observability/plugins/streams_app/public/components/stream_detail_enrichment/sortable_list.tsx
@ -0,0 +1,38 @@
+/*
+ * Copyright Elasticsearch B.V. and/or licensed to Elasticsearch B.V. under one
+ * or more contributor license agreements. Licensed under the Elastic License
+ * 2.0; you may not use this file except in compliance with the Elastic License
+ * 2.0.
+ */
+
+import React from 'react';
+import {
+  DragDropContextProps,
+  EuiDroppableProps,
+  EuiDragDropContext,
+  EuiDroppable,
+  useEuiTheme,
+} from '@elastic/eui';
+import { css } from '@emotion/react';
+
+interface SortableListProps {
+  onDragItem: DragDropContextProps['onDragEnd'];
+  children: EuiDroppableProps['children'];
+}
+
+export const SortableList = ({ onDragItem, children }: SortableListProps) => {
+  const { euiTheme } = useEuiTheme();
+
+  return (
+    <EuiDragDropContext onDragEnd={onDragItem}>
+      <EuiDroppable
+        droppableId="droppable-area"
+        css={css`
+          background-color: ${euiTheme.colors.backgroundTransparent};
+        `}
+      >
+        {children}
+      </EuiDroppable>
+    </EuiDragDropContext>
+  );
+};
--- a/x-pack/solutions/observability/plugins/streams_app/public/components/stream_detail_enrichment/types.ts
+++ b/x-pack/solutions/observability/plugins/streams_app/public/components/stream_detail_enrichment/types.ts
@ -0,0 +1,45 @@
+/*
+ * Copyright Elasticsearch B.V. and/or licensed to Elasticsearch B.V. under one
+ * or more contributor license agreements. Licensed under the Elastic License
+ * 2.0; you may not use this file except in compliance with the Elastic License
+ * 2.0.
+ */
+
+import {
+  DissectProcessingDefinition,
+  FieldDefinitionConfig,
+  GrokProcessingDefinition,
+  ProcessingDefinition,
+} from '@kbn/streams-schema';
+
+export interface ProcessorDefinition extends ProcessingDefinition {
+  id: string;
+}
+
+export interface ProcessingDefinitionGrok extends Pick<ProcessingDefinition, 'condition'> {
+  config: GrokProcessingDefinition;
+}
+
+export interface ProcessingDefinitionDissect extends Pick<ProcessingDefinition, 'condition'> {
+  config: DissectProcessingDefinition;
+}
+
+interface BaseFormState extends Pick<ProcessingDefinition, 'condition'> {
+  detected_fields?: DetectedField[];
+}
+
+export type GrokFormState = BaseFormState &
+  Omit<GrokProcessingDefinition['grok'], 'patterns'> & {
+    type: 'grok';
+    patterns: Array<{ value: string }>;
+  };
+
+export type DissectFormState = DissectProcessingDefinition['dissect'] &
+  BaseFormState & { type: 'dissect' };
+
+export type ProcessorFormState = GrokFormState | DissectFormState;
+
+export interface DetectedField {
+  name: string;
+  type: FieldDefinitionConfig['type'] | 'unmapped';
+}
--- a/x-pack/solutions/observability/plugins/streams_app/public/components/stream_detail_enrichment/utils.ts
+++ b/x-pack/solutions/observability/plugins/streams_app/public/components/stream_detail_enrichment/utils.ts
@ -0,0 +1,137 @@
+/*
+ * Copyright Elasticsearch B.V. and/or licensed to Elasticsearch B.V. under one
+ * or more contributor license agreements. Licensed under the Elastic License
+ * 2.0; you may not use this file except in compliance with the Elastic License
+ * 2.0.
+ */
+
+import {
+  DissectProcessingDefinition,
+  GrokProcessingDefinition,
+  ProcessingDefinition,
+  ProcessorType,
+  isCompleteCondition,
+  isDissectProcessor,
+  isGrokProcessor,
+} from '@kbn/streams-schema';
+import { isEmpty } from 'lodash';
+import { DissectFormState, GrokFormState, ProcessorDefinition, ProcessorFormState } from './types';
+
+const defaultCondition: ProcessingDefinition['condition'] = {
+  field: '',
+  operator: 'eq',
+  value: '',
+};
+
+const defaultGrokProcessorFormState: GrokFormState = {
+  type: 'grok',
+  field: 'message',
+  patterns: [{ value: '' }],
+  pattern_definitions: {},
+  ignore_failure: true,
+  ignore_missing: true,
+  condition: defaultCondition,
+};
+
+const defaultDissectProcessorFormState: DissectFormState = {
+  type: 'dissect',
+  field: 'message',
+  pattern: '',
+  ignore_failure: true,
+  ignore_missing: true,
+  condition: defaultCondition,
+};
+
+const defaultProcessorFormStateByType: Record<ProcessorType, ProcessorFormState> = {
+  dissect: defaultDissectProcessorFormState,
+  grok: defaultGrokProcessorFormState,
+};
+
+export const getDefaultFormState = (
+  type: ProcessorType,
+  processor?: ProcessorDefinition
+): ProcessorFormState => {
+  if (!processor) return defaultProcessorFormStateByType[type];
+
+  let configValues: ProcessorFormState = defaultProcessorFormStateByType[type];
+
+  if (isGrokProcessor(processor.config)) {
+    const { grok } = processor.config;
+
+    configValues = structuredClone({
+      ...grok,
+      type: 'grok',
+      patterns: grok.patterns.map((pattern) => ({ value: pattern })),
+    });
+  }
+
+  if (isDissectProcessor(processor.config)) {
+    const { dissect } = processor.config;
+
+    configValues = structuredClone({
+      ...dissect,
+      type: 'dissect',
+    });
+  }
+
+  return {
+    condition: processor.condition || defaultCondition,
+    ...configValues,
+  };
+};
+
+export const convertFormStateToProcessing = (
+  formState: ProcessorFormState
+): ProcessingDefinition => {
+  if (formState.type === 'grok') {
+    const { condition, patterns, ...grokConfig } = formState;
+
+    return {
+      condition: isCompleteCondition(condition) ? condition : undefined,
+      config: {
+        grok: {
+          patterns: patterns
+            .filter(({ value }) => value.trim().length > 0)
+            .map(({ value }) => value),
+          ...grokConfig,
+        },
+      },
+    };
+  }
+
+  if (formState.type === 'dissect') {
+    const { condition, ...dissectConfig } = formState;
+
+    return {
+      condition: isCompleteCondition(condition) ? condition : undefined,
+      config: {
+        dissect: dissectConfig,
+      },
+    };
+  }
+
+  throw new Error('Cannot convert form state to processing: unknown type.');
+};
+
+export const isCompleteGrokDefinition = (processing: GrokProcessingDefinition) => {
+  const { patterns } = processing.grok;
+
+  return !isEmpty(patterns);
+};
+
+export const isCompleteDissectDefinition = (processing: DissectProcessingDefinition) => {
+  const { pattern } = processing.dissect;
+
+  return !isEmpty(pattern);
+};
+
+export const isCompleteProcessingDefinition = (processing: ProcessingDefinition) => {
+  if (isGrokProcessor(processing.config)) {
+    return isCompleteGrokDefinition(processing.config);
+  }
+  if (isDissectProcessor(processing.config)) {
+    return isCompleteDissectDefinition(processing.config);
+  }
+
+  return false;
+};
--- a/x-pack/solutions/observability/plugins/streams_app/public/components/stream_detail_management/classic.tsx
+++ b/x-pack/solutions/observability/plugins/streams_app/public/components/stream_detail_management/classic.tsx
@ -10,7 +10,7 @@ import { ReadStreamDefinition } from '@kbn/streams-schema';
 import { EuiFlexGroup, EuiListGroup, EuiText } from '@elastic/eui';
 import { useStreamsAppParams } from '../../hooks/use_streams_app_params';
 import { RedirectTo } from '../redirect_to';
-import { StreamDetailEnriching } from '../stream_detail_enriching';
+import { StreamDetailEnrichment } from '../stream_detail_enrichment';
 import { useKibana } from '../../hooks/use_kibana';
 import { Wrapper } from './wrapper';

@ -40,9 +40,9 @@ export function ClassicStreamDetailManagement({
    },
    enrich: {
      content: (
-        <StreamDetailEnriching definition={definition} refreshDefinition={refreshDefinition} />
+        <StreamDetailEnrichment definition={definition} refreshDefinition={refreshDefinition} />
      ),
-      label: i18n.translate('xpack.streams.streamDetailView.enrichingTab', {
+      label: i18n.translate('xpack.streams.streamDetailView.enrichmentTab', {
        defaultMessage: 'Extract field',
      }),
    },
--- a/x-pack/solutions/observability/plugins/streams_app/public/components/stream_detail_management/wired.tsx
+++ b/x-pack/solutions/observability/plugins/streams_app/public/components/stream_detail_management/wired.tsx
@ -10,7 +10,7 @@ import { WiredReadStreamDefinition } from '@kbn/streams-schema';
 import { useStreamsAppParams } from '../../hooks/use_streams_app_params';
 import { RedirectTo } from '../redirect_to';
 import { StreamDetailRouting } from '../stream_detail_routing';
-import { StreamDetailEnriching } from '../stream_detail_enriching';
+import { StreamDetailEnrichment } from '../stream_detail_enrichment';
 import { StreamDetailSchemaEditor } from '../stream_detail_schema_editor';
 import { Wrapper } from './wrapper';

@ -44,9 +44,9 @@ export function WiredStreamDetailManagement({
    },
    enrich: {
      content: (
-        <StreamDetailEnriching definition={definition} refreshDefinition={refreshDefinition} />
+        <StreamDetailEnrichment definition={definition} refreshDefinition={refreshDefinition} />
      ),
-      label: i18n.translate('xpack.streams.streamDetailView.enrichingTab', {
+      label: i18n.translate('xpack.streams.streamDetailView.enrichmentTab', {
        defaultMessage: 'Extract field',
      }),
    },
--- a/x-pack/solutions/observability/plugins/streams_app/public/components/stream_detail_management/wrapper.tsx
+++ b/x-pack/solutions/observability/plugins/streams_app/public/components/stream_detail_management/wrapper.tsx
@ -23,7 +23,7 @@ export function Wrapper({
  return (
    <EuiFlexGroup
      direction="column"
-      gutterSize="s"
+      gutterSize="l"
      className={css`
        max-width: 100%;
      `}
--- a/x-pack/solutions/observability/plugins/streams_app/public/components/stream_detail_overview/index.tsx
+++ b/x-pack/solutions/observability/plugins/streams_app/public/components/stream_detail_overview/index.tsx
@ -8,7 +8,6 @@ import {
  EuiButton,
  EuiFlexGroup,
  EuiFlexItem,
-  EuiImage,
  EuiLoadingSpinner,
  EuiPanel,
  EuiTab,
@ -22,7 +21,6 @@ import React, { useMemo } from 'react';
 import { css } from '@emotion/css';
 import { ReadStreamDefinition, isWiredReadStream, isWiredStream } from '@kbn/streams-schema';
 import { useDateRange } from '@kbn/observability-utils-browser/hooks/use_date_range';
-import illustration from '../assets/illustration.png';
 import { useKibana } from '../../hooks/use_kibana';
 import { useStreamsAppFetch } from '../../hooks/use_streams_app_fetch';
 import { ControlledEsqlChart } from '../esql_chart/controlled_esql_chart';
@ -30,6 +28,7 @@ import { StreamsAppSearchBar } from '../streams_app_search_bar';
 import { getIndexPatterns } from '../../util/hierarchy_helpers';
 import { StreamsList } from '../streams_list';
 import { useStreamsAppRouter } from '../../hooks/use_streams_app_router';
+import { AssetImage } from '../asset_image';

 const formatNumber = (val: number) => {
  return Number(val).toLocaleString('en', {
@ -303,13 +302,7 @@ function ChildStreamList({ stream }: { stream?: ReadStreamDefinition }) {
            `}
          >
            <EuiFlexGroup direction="column" gutterSize="s">
-              <EuiImage
-                src={illustration}
-                alt="Illustration"
-                className={css`
-                  width: 250px;
-                `}
-              />
+              <AssetImage type="welcome" />
              <EuiText size="m" textAlign="center">
                {i18n.translate('xpack.streams.entityDetailOverview.noChildStreams', {
                  defaultMessage: 'Create streams for your logs',
--- a/x-pack/solutions/observability/plugins/streams_app/public/components/stream_detail_routing/index.tsx
+++ b/x-pack/solutions/observability/plugins/streams_app/public/components/stream_detail_routing/index.tsx
@ -13,7 +13,6 @@ import {
  EuiFlexItem,
  EuiFormRow,
  EuiIcon,
-  EuiImage,
  EuiLoadingSpinner,
  EuiPanel,
  EuiResizableContainer,
@ -39,9 +38,9 @@ import { ConditionEditor } from '../condition_editor';
 import { useDebounced } from '../../util/use_debounce';
 import { useStreamsAppRouter } from '../../hooks/use_streams_app_router';
 import { NestedView } from '../nested_view';
-import illustration from '../assets/illustration.png';
 import { PreviewTable } from '../preview_table';
 import { StreamDeleteModal } from '../stream_delete_modal';
+import { AssetImage } from '../asset_image';

 function useRoutingState() {
  const [childUnderEdit, setChildUnderEdit] = React.useState<
@ -463,13 +462,7 @@ function PreviewPanelIllustration({
          max-width: 350px;
        `}
      >
-        <EuiImage
-          src={illustration}
-          alt="Illustration"
-          className={css`
-            width: 250px;
-          `}
-        />
+        <AssetImage />
        {previewSampleFetch.loading ? (
          <EuiText size="xs" textAlign="center">
            <EuiLoadingSpinner size="s" />
--- a/x-pack/solutions/observability/plugins/streams_app/public/components/streams_app_search_bar/index.tsx
+++ b/x-pack/solutions/observability/plugins/streams_app/public/components/streams_app_search_bar/index.tsx
@ -10,7 +10,7 @@ import React, { useMemo } from 'react';
 import type { DataView } from '@kbn/data-views-plugin/common';
 import { useKibana } from '../../hooks/use_kibana';

-interface Props {
+export interface StreamsAppSearchBarProps {
  query?: string;
  dateRangeFrom?: string;
  dateRangeTo?: string;
@ -30,7 +30,7 @@ export function StreamsAppSearchBar({
  query,
  placeholder,
  dataViews,
-}: Props) {
+}: StreamsAppSearchBarProps) {
  const {
    dependencies: {
      start: { unifiedSearch },
--- a/x-pack/solutions/observability/plugins/streams_app/public/hooks/use_discard_confirm.ts
+++ b/x-pack/solutions/observability/plugins/streams_app/public/hooks/use_discard_confirm.ts
@ -0,0 +1,35 @@
+/*
+ * Copyright Elasticsearch B.V. and/or licensed to Elasticsearch B.V. under one
+ * or more contributor license agreements. Licensed under the Elastic License
+ * 2.0; you may not use this file except in compliance with the Elastic License
+ * 2.0.
+ */
+
+import { i18n } from '@kbn/i18n';
+import { useKibana } from './use_kibana';
+
+export const useDiscardConfirm = (handler: () => void) => {
+  const { core } = useKibana();
+
+  return async () => {
+    const hasCancelled = await core.overlays.openConfirm(
+      i18n.translate('xpack.streams.cancelModal.message', {
+        defaultMessage: 'Are you sure you want to discard your changes?',
+      }),
+      {
+        buttonColor: 'danger',
+        title: i18n.translate('xpack.streams.cancelModal.title', {
+          defaultMessage: 'Discard changes?',
+        }),
+        confirmButtonText: i18n.translate('xpack.streams.cancelModal.confirm', {
+          defaultMessage: 'Discard',
+        }),
+        cancelButtonText: i18n.translate('xpack.streams.cancelModal.cancel', {
+          defaultMessage: 'Keep editing',
+        }),
+      }
+    );
+
+    if (hasCancelled) handler();
+  };
+};
--- a/x-pack/solutions/observability/plugins/streams_app/public/hooks/use_kibana.tsx
+++ b/x-pack/solutions/observability/plugins/streams_app/public/hooks/use_kibana.tsx
@ -5,13 +5,14 @@
 * 2.0.
 */

-import type { CoreStart } from '@kbn/core/public';
+import type { AppMountParameters, CoreStart } from '@kbn/core/public';
 import { useMemo } from 'react';
 import { useKibana } from '@kbn/kibana-react-plugin/public';
 import type { StreamsAppStartDependencies } from '../types';
 import type { StreamsAppServices } from '../services/types';

 export interface StreamsAppKibanaContext {
+  appParams: AppMountParameters;
  core: CoreStart;
  dependencies: {
    start: StreamsAppStartDependencies;
@ -23,9 +24,10 @@ const useTypedKibana = (): StreamsAppKibanaContext => {
  const context = useKibana<CoreStart & Omit<StreamsAppKibanaContext, 'core'>>();

  return useMemo(() => {
-    const { dependencies, services, ...core } = context.services;
+    const { appParams, dependencies, services, ...core } = context.services;

    return {
+      appParams,
      core,
      dependencies,
      services,
--- a/x-pack/solutions/observability/plugins/streams_app/tsconfig.json
+++ b/x-pack/solutions/observability/plugins/streams_app/tsconfig.json
@ -48,5 +48,11 @@
    "@kbn/core-notifications-browser",
    "@kbn/index-lifecycle-management-common-shared",
    "@kbn/streams-schema",
+    "@kbn/react-hooks",
+    "@kbn/i18n-react",
+    "@kbn/react-field",
+    "@kbn/shared-ux-utility",
+    "@kbn/unsaved-changes-prompt",
+    "@kbn/object-utils"
  ]
 }
--- a/yarn.lock
+++ b/yarn.lock
@ -6476,6 +6476,10 @@
  version "0.0.0"
  uid ""

+"@kbn/object-utils@link:src/platform/packages/shared/kbn-object-utils":
+  version "0.0.0"
+  uid ""
+
 "@kbn/object-versioning-utils@link:src/platform/packages/shared/kbn-object-versioning-utils":
  version "0.0.0"
  uid ""