🌊 Streams: Normalized format for wired streams

x-pack/platform/packages/shared/kbn-streams-schema/index.ts

@@ -47,6 +47,12 @@ export { type RoutingDefinition, routingDefinitionListSchema } from './src/model
 export { type ContentPack, contentPackSchema } from './src/content';
 
 export { isRootStreamDefinition } from './src/helpers/is_root';
+export {
+  keepFields,
+  namespacePrefixes,
+  isNamespacedEcsField,
+  getRegularEcsField,
+} from './src/helpers/namespaced_ecs';
 export { getAdvancedParameters } from './src/helpers/get_advanced_parameters';
 export { getInheritedFieldsFromAncestors } from './src/helpers/get_inherited_fields_from_ancestors';
 
@@ -73,6 +79,7 @@ export {
   type NamedFieldDefinitionConfig,
   type FieldDefinitionConfig,
   type InheritedFieldDefinitionConfig,
+  type InheritedFieldDefinition,
   type FieldDefinitionConfigAdvancedParameters,
   fieldDefinitionConfigSchema,
   namedFieldDefinitionConfigSchema,

x-pack/platform/packages/shared/kbn-streams-schema/src/fields/index.ts

@@ -62,6 +62,7 @@ export const fieldDefinitionSchema: z.Schema<FieldDefinition> = z.record(
 
 export type InheritedFieldDefinitionConfig = FieldDefinitionConfig & {
   from: string;
+  alias_for?: string;
 };
 
 export interface InheritedFieldDefinition {
@@ -70,7 +71,10 @@ export interface InheritedFieldDefinition {
 
 export const inheritedFieldDefinitionSchema: z.Schema<InheritedFieldDefinition> = z.record(
   z.string(),
-  z.intersection(fieldDefinitionConfigSchema, z.object({ from: NonEmptyString }))
+  z.intersection(
+    fieldDefinitionConfigSchema,
+    z.object({ from: NonEmptyString, alias_for: z.optional(NonEmptyString) })
+  )
 );
 
 export type NamedFieldDefinitionConfig = FieldDefinitionConfig & {

x-pack/platform/packages/shared/kbn-streams-schema/src/helpers/namespaced_ecs.ts

@@ -0,0 +1,55 @@
+/*
+ * 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 const keepFields: string[] = [
+  '@timestamp',
+  'observed_timestamp',
+  'trace_id',
+  'span_id',
+  'severity_text',
+  'body',
+  'severity_number',
+  'event_name',
+  'dropped_attributes_count',
+  'scope',
+  'body.text',
+  'body.structured',
+  'resource.schema_url',
+  'resource.dropped_attributes_count',
+];
+
+export const aliases: Record<string, string> = {
+  trace_id: 'trace.id',
+  span_id: 'span.id',
+  severity_text: 'log.level',
+  'body.text': 'message',
+};
+
+export const namespacePrefixes = [
+  'body.structured.',
+  'attributes.',
+  'scope.attributes.',
+  'resource.attributes.',
+];
+
+export function getRegularEcsField(field: string): string {
+  // check whether it starts with a namespace prefix
+  for (const prefix of namespacePrefixes) {
+    if (field.startsWith(prefix)) {
+      return field.slice(prefix.length);
+    }
+  }
+  // check aliases
+  if (aliases[field]) {
+    return aliases[field];
+  }
+  return field;
+}
+
+export function isNamespacedEcsField(field: string): boolean {
+  return namespacePrefixes.some((prefix) => field.startsWith(prefix)) || keepFields.includes(field);
+}

x-pack/platform/plugins/shared/streams/server/lib/streams/component_templates/generate_layer.test.ts

@@ -0,0 +1,188 @@
+/*
+ * 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 { Streams } from '@kbn/streams-schema';
+import { generateLayer } from './generate_layer';
+
+describe('generateLayer', () => {
+  const definition: Streams.WiredStream.Definition = {
+    name: 'logs.abc',
+    description: '',
+    ingest: {
+      processing: [],
+      wired: {
+        routing: [],
+        fields: {
+          '@timestamp': { type: 'date', format: 'strict_date_optional_time' },
+          message: { type: 'match_only_text' },
+          'attributes.myfield': { type: 'keyword' },
+        },
+      },
+      lifecycle: {
+        // simulate DSL lifecycle
+        dsl: { data_retention: '30d' },
+      },
+    },
+  };
+
+  it('should generate mappings with proper handling of fields and aliases', () => {
+    const result = generateLayer('logs.abc', definition, false);
+    expect(result).toMatchInlineSnapshot(`
+      Object {
+        "_meta": Object {
+          "description": "Default settings for the logs.abc stream",
+          "managed": true,
+        },
+        "name": "logs.abc@stream.layer",
+        "template": Object {
+          "lifecycle": Object {
+            "data_retention": "30d",
+            "enabled": true,
+          },
+          "mappings": Object {
+            "dynamic": false,
+            "properties": Object {
+              "@timestamp": Object {
+                "format": "strict_date_optional_time",
+                "ignore_malformed": false,
+                "type": "date",
+              },
+              "attributes.myfield": Object {
+                "type": "keyword",
+              },
+              "message": Object {
+                "type": "match_only_text",
+              },
+              "myfield": Object {
+                "path": "attributes.myfield",
+                "type": "alias",
+              },
+            },
+          },
+          "settings": Object {
+            "index.lifecycle.name": undefined,
+            "index.lifecycle.prefer_ilm": false,
+          },
+        },
+        "version": 1,
+      }
+    `);
+  });
+
+  it('should generate mappings for root stream', () => {
+    const result = generateLayer('logs', { ...definition, name: 'logs' }, true);
+    expect(result).toMatchInlineSnapshot(`
+      Object {
+        "_meta": Object {
+          "description": "Default settings for the logs stream",
+          "managed": true,
+        },
+        "name": "logs@stream.layer",
+        "template": Object {
+          "lifecycle": Object {
+            "data_retention": "30d",
+          },
+          "mappings": Object {
+            "dynamic": false,
+            "properties": Object {
+              "@timestamp": Object {
+                "format": "strict_date_optional_time",
+                "ignore_malformed": false,
+                "type": "date",
+              },
+              "attributes": Object {
+                "subobjects": false,
+                "type": "object",
+              },
+              "attributes.myfield": Object {
+                "type": "keyword",
+              },
+              "body": Object {
+                "properties": Object {
+                  "structured": Object {
+                    "type": "flattened",
+                  },
+                  "text": Object {
+                    "type": "match_only_text",
+                  },
+                },
+                "type": "object",
+              },
+              "log.level": Object {
+                "path": "severity_text",
+                "type": "alias",
+              },
+              "message": Object {
+                "type": "match_only_text",
+              },
+              "myfield": Object {
+                "path": "attributes.myfield",
+                "type": "alias",
+              },
+              "resource": Object {
+                "properties": Object {
+                  "attributes": Object {
+                    "subobjects": false,
+                    "type": "object",
+                  },
+                  "dropped_attributes_count": Object {
+                    "type": "long",
+                  },
+                  "schema_url": Object {
+                    "ignore_above": 1024,
+                    "type": "keyword",
+                  },
+                },
+                "type": "object",
+              },
+              "scope": Object {
+                "properties": Object {
+                  "attributes": Object {
+                    "subobjects": false,
+                    "type": "object",
+                  },
+                },
+                "type": "object",
+              },
+              "span.id": Object {
+                "path": "span_id",
+                "type": "alias",
+              },
+              "trace.id": Object {
+                "path": "trace_id",
+                "type": "alias",
+              },
+            },
+          },
+          "settings": Object {
+            "index": Object {
+              "codec": "best_compression",
+              "mapping": Object {
+                "ignore_malformed": true,
+                "total_fields": Object {
+                  "ignore_dynamic_beyond_limit": true,
+                },
+              },
+              "mode": "logsdb",
+              "sort": Object {
+                "field": Array [
+                  "resource.attributes.host.name",
+                  "@timestamp",
+                ],
+                "order": Array [
+                  "asc",
+                  "desc",
+                ],
+              },
+            },
+          },
+        },
+        "version": 1,
+      }
+    `);
+  });
+});

x-pack/platform/plugins/shared/streams/server/lib/streams/component_templates/generate_layer.ts

@@ -16,10 +16,12 @@ import {
   isDslLifecycle,
   isIlmLifecycle,
   isRoot,
+  namespacePrefixes,
 } from '@kbn/streams-schema';
 import { ASSET_VERSION } from '../../../../common/constants';
 import { logsSettings } from './logs_layer';
 import { getComponentTemplateName } from './name';
+import { baseMappings } from './logs_layer';
 
 export function generateLayer(
   name: string,
@@ -50,6 +52,14 @@ export function generateLayer(
     }
 
     properties[field] = property;
+    const matchingPrefix = namespacePrefixes.find((prefix) => field.startsWith(prefix));
+    if (matchingPrefix) {
+      const aliasName = field.substring(matchingPrefix.length);
+      properties[aliasName] = {
+        type: 'alias',
+        path: field,
+      };
+    }
   });
 
   return {
@@ -58,9 +68,13 @@ export function generateLayer(
       lifecycle: getTemplateLifecycle(definition, isServerless),
       settings: getTemplateSettings(definition, isServerless),
       mappings: {
-        subobjects: false,
         dynamic: false,
-        properties,
+        properties: isRoot(name)
+          ? {
+              ...baseMappings,
+              ...properties,
+            }
+          : properties,
       },
     },
     version: ASSET_VERSION,