[Streams] Consistent Grok pattern generation (#230076)

Resolves elastic/streams-program#314

## Summary 

Updates the Grok pattern suggestions feature to use heuristics when
constructing Grok patterns in order to improve reliability. Once a Grok
pattern has been created an LLM is used to determine ECS field names.

Introduces a new shared package `@kbn/grok-heuristics` which exposes
Grok pattern extraction and grouping utilities for log/message analysis.

## Acceptance criteria

- Generate pattern button should create one GROK processor suggestion
showing parse rate and field metrics
- The user can accept the suggestion or dismiss it
- When accepting the suggestion any existing patterns and pattern
definitions are replaced with the suggested grok processor
- When multiple log formats are detected in the dataset the suggested
processor should contain a list of fallback patterns, one for each log
format. The maximum number of patterns returned is limited to 3.
- For each suggested Grok pattern the UI should call the selected LLM to
suggest field names and merge any fields that have been broken up too
granular
- The Grok pattern should extract fields to ECS convention for classic
streams or Otel convention for wired streams. Custom timestamps should
never be extracted to `@timestamp` to avoid date format conflicts.
- Fields that span at least 2 Grok components should be extracted into
their own pattern definition
- When multiple conflicting pattern definitions are returned by the LLM
should resolve them by appending a counter (e.g. CUSTOM_TIMESTAMP2 for
the second pattern definition)

## Screenshot

<img width="1363" height="1086" alt="Screenshot 2025-08-13 at 08 39 48"
src="https://github.com/user-attachments/assets/71791309-a2b0-4769-8ffd-d3fa3ea11c74"
/>


## Evaluations

| Stream            | Before (all docs) | After (all docs) |

|--------------------|-------------------------|------------------------|
| `logs.android`     | 0%                     | 100%                |
| `logs.apache`      | 100%                   | 100%                  |
| `logs.bgl`         | 0%                     | 100%                |
| `logs.hadoop`      | 100%                   | 100%                  |
| `logs.hdfs`        | 0%                     | 100%                |
| `logs.healthapp`   | 100%                   | 100%                  |
| `logs.hpc`         | 100%                   | 100%                  |
| `logs.linux`       | 100%                   | 100%                  |
| `logs.mac`         | 100%                   | 100%                  |
| `logs.openssh`     | 71.9%                  | 100%                |
| `logs.openstack`   | 100%                   | 100%                  |
| `logs.proxifier`   | 51.0%                  | 99.8%               |
| `logs.spark`       | 99.6%                  | 99.4%                 |
| `logs.thunderbird` | 58.1%                  | 95.2%               |
| `logs.windows`     | 100%                   | 100%                  |
| `logs.zookeeper`   | 100%                   | 100%                  |

| Metric                        | Before | After  |
|-------------------------------|--------|--------|
| Average Parsing Score (samples) | 75.2% | 100% |
| Average Parsing Score (all docs) | 73.8% | 99.7% |

---------

Co-authored-by: Dario Gieselaar <[email protected]>
Co-authored-by: kibanamachine <[email protected]>

Author: 3 people

.github/CODEOWNERS

@@ -889,6 +889,7 @@ x-pack/platform/packages/shared/kbn-entities-schema @elastic/obs-entities
 x-pack/platform/packages/shared/kbn-evals @elastic/appex-ai-infra
 x-pack/platform/packages/shared/kbn-event-stacktrace @elastic/obs-ux-infra_services-team @elastic/obs-ux-logs-team
 x-pack/platform/packages/shared/kbn-failure-store-modal @elastic/kibana-management
+x-pack/platform/packages/shared/kbn-grok-heuristics/package.json @elastic/streams-program-team
 x-pack/platform/packages/shared/kbn-inference-cli @elastic/appex-ai-infra
 x-pack/platform/packages/shared/kbn-inference-endpoint-ui-common @elastic/appex-ai-infra
 x-pack/platform/packages/shared/kbn-inference-prompt-utils @elastic/appex-ai-infra

package.json

@@ -595,6 +595,7 @@
     "@kbn/graph-plugin": "link:x-pack/platform/plugins/private/graph",
     "@kbn/grid-example-plugin": "link:examples/grid_example",
     "@kbn/grid-layout": "link:src/platform/packages/private/kbn-grid-layout",
+    "@kbn/grok-heuristics": "link:x-pack/platform/packages/shared/kbn-grok-heuristics",
     "@kbn/grok-ui": "link:src/platform/packages/shared/kbn-grok-ui",
     "@kbn/grokdebugger-plugin": "link:x-pack/platform/plugins/private/grokdebugger",
     "@kbn/grouping": "link:src/platform/packages/shared/kbn-grouping",

tsconfig.base.json

@@ -1110,6 +1110,8 @@
       "@kbn/grid-example-plugin/*": ["examples/grid_example/*"],
       "@kbn/grid-layout": ["src/platform/packages/private/kbn-grid-layout"],
       "@kbn/grid-layout/*": ["src/platform/packages/private/kbn-grid-layout/*"],
+      "@kbn/grok-heuristics": ["x-pack/platform/packages/shared/kbn-grok-heuristics"],
+      "@kbn/grok-heuristics/*": ["x-pack/platform/packages/shared/kbn-grok-heuristics/*"],
       "@kbn/grok-ui": ["src/platform/packages/shared/kbn-grok-ui"],
       "@kbn/grok-ui/*": ["src/platform/packages/shared/kbn-grok-ui/*"],
       "@kbn/grokdebugger-plugin": ["x-pack/platform/plugins/private/grokdebugger"],
@@ -2342,13 +2344,23 @@
       "@kbn/zod-helpers/*": ["src/platform/packages/shared/kbn-zod-helpers/*"],
       // END AUTOMATED PACKAGE LISTING
       // Allows for importing from `kibana` package for the exported types.
-      "@emotion/core": ["typings/@emotion"],
+      "@emotion/core": [
+        "typings/@emotion"
+      ],
       // We need the custom typings "proxy" because our current import-resolver doesn't support "exports" in package.json.
       // We should be able to remove this once we support cjs/esm interop.
-      "@elastic/opentelemetry-node/sdk": ["typings/@elastic/opentelemetry-node/sdk"],
-      "@opentelemetry/semantic-conventions/incubating": ["typings/@opentelemetry/semantic-conventions/incubating"],
-      "@typescript-eslint/parser": ["typings/@typescript-eslint/parser"],
-      "@a2a-js/sdk/server": ["typings/@a2a-js-sdk/server"],
+      "@elastic/opentelemetry-node/sdk": [
+        "typings/@elastic/opentelemetry-node/sdk"
+      ],
+      "@opentelemetry/semantic-conventions/incubating": [
+        "typings/@opentelemetry/semantic-conventions/incubating"
+      ],
+      "@typescript-eslint/parser": [
+        "typings/@typescript-eslint/parser"
+      ],
+      "@a2a-js/sdk/server": [
+        "typings/@a2a-js-sdk/server"
+      ],
     },
     // Support .tsx files and transform JSX into calls to React.createElement
     "jsx": "react",

…treams/scripts/evaluate_grok_patterns.ts …ipts/evaluate_heuristic_grok_patterns.tsx-pack/platform/plugins/shared/streams/scripts/evaluate_grok_patterns.ts renamed to x-pack/platform/packages/private/kbn-evals-suite-streams/scripts/evaluate_heuristic_grok_patterns.ts x-pack/platform/plugins/shared/streams/scripts/evaluate_grok_patterns.ts renamed to x-pack/platform/packages/private/kbn-evals-suite-streams/scripts/evaluate_heuristic_grok_patterns.ts

@@ -17,20 +17,24 @@
  * Run evaluation script using:
  *
  * ```bash
- * yarn run ts-node --transpile-only x-pack/platform/plugins/shared/streams/scripts/evaluate_grok_patterns.ts
+ * node --require ./src/setup_node_env/ ./x-pack/platform/plugins/shared/streams/scripts/evaluate_heuristic_grok_patterns.ts
  * ```
  */
 
 import { Client } from '@elastic/elasticsearch';
 import fetch from 'node-fetch';
-import uniq from 'lodash/uniq';
 import { writeFile } from 'fs/promises';
 import { join } from 'path';
 import chalk from 'chalk';
 import yargs from 'yargs/yargs';
 import { flattenObject } from '@kbn/object-utils';
 import { get } from 'lodash';
-import { getLogGroups } from '../server/routes/internal/streams/processing/get_log_groups';
+import {
+  getReviewFields,
+  getGrokProcessor,
+  getGrokPattern,
+  extractGrokPatternDangerouslySlow,
+} from '@kbn/grok-heuristics';
 
 const ES_URL = 'http://localhost:9200';
 const ES_USER = 'elastic';
@@ -72,11 +76,11 @@ async function getStreams(): Promise<string[]> {
   const data = await fetch(`${KIBANA_URL}/api/streams`, {
     method: 'GET',
     headers: getKibanaAuthHeaders(),
-  }).then((res) => {
+  }).then(async (res) => {
     if (res.ok) {
       return res.json();
     }
-    throw new Error(`HTTP Response Code: ${res.status}`);
+    throw new Error(`HTTP Response (${res.status}): ${await res.text()}`);
   });
   return data.streams.map((s: any) => s.name).filter((name: string) => name.startsWith('logs.'));
 }
@@ -85,29 +89,37 @@ async function getConnectors(): Promise<string[]> {
   const data = await fetch(`${KIBANA_URL}/api/actions/connectors`, {
     method: 'GET',
     headers: getKibanaAuthHeaders(),
-  }).then((res) => {
+  }).then(async (res) => {
     if (res.ok) {
       return res.json();
     }
-    throw new Error(`HTTP Response Code: ${res.status}`);
+    throw new Error(`HTTP Response (${res.status}): ${await res.text()}`);
   });
   return data.map((c: any) => c.id);
 }
 
-async function getSuggestions(stream: string, connectorId: string, samples: any[]) {
-  const data = await fetch(`${KIBANA_URL}/internal/streams/${stream}/processing/_suggestions`, {
-    method: 'POST',
-    headers: getKibanaAuthHeaders(),
-    body: JSON.stringify({
-      connectorId,
-      field: MESSAGE_FIELD,
-      samples,
-    }),
-  }).then(async (res) => {
+async function getSuggestions(
+  stream: string,
+  connectorId: string,
+  messages: string[],
+  reviewFields: ReturnType<typeof getReviewFields>
+) {
+  const data = await fetch(
+    `${KIBANA_URL}/internal/streams/${stream}/processing/_suggestions/grok`,
+    {
+      method: 'POST',
+      headers: getKibanaAuthHeaders(),
+      body: JSON.stringify({
+        connector_id: connectorId,
+        sample_messages: messages.slice(0, 10),
+        review_fields: reviewFields,
+      }),
+    }
+  ).then(async (res) => {
     if (res.ok) {
       return res.json();
     }
-    throw new Error(`HTTP Response Code: ${res.status}`);
+    throw new Error(`HTTP Response (${res.status}): ${await res.text()}`);
   });
   return data;
 }
@@ -118,24 +130,25 @@ async function simulateGrokProcessor(stream: string, documents: any[], grokProce
     headers: getKibanaAuthHeaders(),
     body: JSON.stringify({
       documents,
-      processing: {
-        steps: [
-          {
-            customIdentifier: 'eval-grok',
-            from: grokProcessor.field,
+      processing: [
+        {
+          id: 'eval-grok',
+          grok: {
+            field: MESSAGE_FIELD,
             patterns: grokProcessor.patterns,
+            pattern_definitions: grokProcessor.pattern_definitions,
             ignore_failure: true,
             ignore_missing: true,
-            where: { always: {} },
+            if: { always: {} },
           },
-        ],
-      },
+        },
+      ],
     }),
-  }).then((res) => {
+  }).then(async (res) => {
     if (res.ok) {
       return res.json();
     }
-    throw new Error(`HTTP Response Code: ${res.status}`);
+    throw new Error(`HTTP Response (${res.status}): ${await res.text()}`);
   });
   return data;
 }
@@ -172,7 +185,7 @@ export async function evaluateGrokSuggestions() {
     );
   }
 
-  const connector = connectors[connectors.length - 1]; // Use the last connector for evaluation
+  const connector = connectors[0]; // Use the first connector for evaluation
 
   // 1. Get AI suggestions
   console.log(chalk.bold('Getting suggestions...'));
@@ -181,12 +194,25 @@ export async function evaluateGrokSuggestions() {
     streams.map(async (stream) => {
       const sampleDocs = await fetchDocs(stream, 100);
 
-      const suggestion = await getSuggestions(stream, connector, sampleDocs);
-      const grokProcessor = suggestion[0]?.grokProcessor;
+      const messages = sampleDocs.reduce<string[]>((acc, sample) => {
+        const value = get(sample, MESSAGE_FIELD);
+        if (typeof value === 'string') {
+          acc.push(value);
+        }
+        return acc;
+      }, []);
+
+      const grokPatternNodes = extractGrokPatternDangerouslySlow(messages);
+      const grokPattern = getGrokPattern(grokPatternNodes);
+      const reviewFields = getReviewFields(grokPatternNodes, 10);
+      console.log(`- ${stream}: ${chalk.dim(grokPattern)}`);
+
+      const suggestion = await getSuggestions(stream, connector, messages, reviewFields);
+      const grokProcessor = getGrokProcessor(grokPatternNodes, suggestion);
       if (!grokProcessor) {
         throw new Error('No grokProcessor returned');
       }
-      console.log(`- ${stream}: ${chalk.dim(grokProcessor.patterns.join(', '))}`);
+      console.log(`- ${stream}: ${chalk.green(grokProcessor.patterns.join(', '))}`);
 
       return { stream, ...grokProcessor };
     })
@@ -232,7 +258,7 @@ export async function evaluateGrokSuggestions() {
     chalk.bold(`Average Parsing Score (all docs): ${chalk.green(averageParsingScoreAllDocs)}`)
   );
 
-  return output.reduce((acc, suggestion) => {
+  return output.reduce<Record<string, any>>((acc, suggestion) => {
     acc[suggestion.stream] = {
       patterns: suggestion.patterns,
       pattern_definitions: suggestion.pattern_definitions,
@@ -243,29 +269,6 @@ export async function evaluateGrokSuggestions() {
   }, {});
 }
 
-export async function evaluateLogGrouping() {
-  const allDocs = await fetchDocs('logs.*', 10_000);
-  const groups = getLogGroups(
-    allDocs.map((doc) => `${get(doc, MESSAGE_FIELD)}|||${get(doc, 'attributes.filepath')}`),
-    1
-  );
-  const output = groups.map((g) => {
-    return {
-      pattern: g.pattern,
-      logs: g.logs.length,
-      streams: uniq(g.logs.map((log) => log.split('|||')[1])),
-    };
-  });
-  output.forEach((g) => {
-    console.log();
-    console.log(chalk.bold(`"${g.pattern}" (${g.logs} logs):`));
-    g.streams.forEach((stream) => {
-      console.log(`- ${chalk.green(stream)}`);
-    });
-  });
-  return output;
-}
-
 async function runGrokSuggestionsEvaluation() {
   await evaluateGrokSuggestions()
     .then((result) => {
@@ -278,20 +281,6 @@ async function runGrokSuggestionsEvaluation() {
     .catch(console.error);
 }
 
-async function runLogGroupingEvaluation() {
-  console.log();
-  console.log('Starting evaluation of Log Grouping...');
-  await evaluateLogGrouping()
-    .then((result) => {
-      const file = `grouping_results.${Date.now()}.json`;
-      console.log();
-      console.log(`Evaluation complete. Writing results to ${file}`);
-      return writeFile(join(__dirname, file), JSON.stringify(result, null, 2));
-    })
-    .catch(console.error);
-}
-
 yargs(process.argv.slice(2))
   .command('*', 'Evaluate AI suggestions for Grok patterns', runGrokSuggestionsEvaluation)
-  .command('grouping', 'Evaluate log grouping patterns', runLogGroupingEvaluation)
   .parse();

x-pack/platform/packages/private/kbn-evals-suite-streams/tsconfig.json

@@ -9,5 +9,7 @@
   "kbn_references": [
     "@kbn/evals",
     "@kbn/sse-utils-client",
+    "@kbn/object-utils",
+    "@kbn/grok-heuristics",
   ]
 }

x-pack/platform/packages/shared/kbn-grok-heuristics/README.md

@@ -0,0 +1,3 @@
+# @kbn/grok-heuristics
+
+Utilities and helper functions for extracting GROK patterns from log messages.

x-pack/platform/packages/shared/kbn-grok-heuristics/index.ts

@@ -0,0 +1,15 @@
+/*
+ * 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.
+ */
+
+export { getReviewFields } from './src/review/get_review_fields';
+export { getGrokPattern } from './src/review/get_grok_pattern';
+export { unwrapPatternDefinitions } from './src/review/unwrap_pattern_definitions';
+export { getGrokProcessor, type GrokProcessorResult } from './src/review/get_grok_processor';
+export { mergeGrokProcessors } from './src/review/merge_grok_processors';
+export { groupMessagesByPattern } from './src/group_messages';
+export { extractGrokPatternDangerouslySlow } from './src/tokenization/extract_grok_pattern';
+export { ReviewFieldsPrompt } from './src/review/review_fields_prompt';

x-pack/platform/packages/shared/kbn-grok-heuristics/jest.config.js

@@ -0,0 +1,12 @@
+/*
+ * 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.
+ */
+
+module.exports = {
+  preset: '@kbn/test/jest_node',
+  rootDir: '../../../../..',
+  roots: ['<rootDir>/x-pack/platform/packages/shared/kbn-grok-heuristics'],
+};

x-pack/platform/packages/shared/kbn-grok-heuristics/kibana.jsonc

@@ -0,0 +1,7 @@
+{
+  "type": "shared-common",
+  "id": "@kbn/grok-heuristics",
+  "owner": "@elastic/streams-program-team",
+  "group": "platform",
+  "visibility": "shared"
+}

x-pack/platform/packages/shared/kbn-grok-heuristics/package.json

@@ -0,0 +1,6 @@
+{
+  "name": "@kbn/grok-heuristics",
+  "private": true,
+  "version": "1.0.0",
+  "license": "Elastic License 2.0"
+}