[Observability Onboarding] Wired streams onboarding

x-pack/platform/plugins/shared/fleet/server/services/epm/package_service.ts

@@ -143,7 +143,8 @@ export interface PackageClient {
     pkgVersion?: string,
     isInputIncluded?: (input: TemplateAgentPolicyInput) => boolean,
     prerelease?: boolean,
-    ignoreUnverified?: boolean
+    ignoreUnverified?: boolean,
+    injectWiredStreamsRouting?: boolean
   ): Promise<string>;
 
   reinstallEsAssets(
@@ -326,7 +327,8 @@ class PackageClientImpl implements PackageClient {
     pkgVersion?: string,
     isInputIncluded?: (input: TemplateAgentPolicyInput) => boolean,
     prerelease?: boolean,
-    ignoreUnverified?: boolean
+    ignoreUnverified?: boolean,
+    injectWiredStreamsRouting?: boolean
   ) {
     await this.#runPreflight(READ_PACKAGE_INFO_AUTHZ);
 
@@ -343,7 +345,8 @@ class PackageClientImpl implements PackageClient {
       'yml',
       isInputIncluded,
       prerelease,
-      ignoreUnverified
+      ignoreUnverified,
+      injectWiredStreamsRouting
     );
   }
 

x-pack/platform/plugins/shared/fleet/server/services/epm/packages/get_template_inputs.ts

@@ -100,7 +100,8 @@ export async function getTemplateInputs(
   format: 'yml',
   isInputIncluded?: (input: TemplateAgentPolicyInput) => boolean,
   prerelease?: boolean,
-  ignoreUnverified?: boolean
+  ignoreUnverified?: boolean,
+  injectWiredStreamsRouting?: boolean
 ): Promise<string>;
 export async function getTemplateInputs(
   soClient: SavedObjectsClientContract,
@@ -109,7 +110,8 @@ export async function getTemplateInputs(
   format: 'json',
   isInputIncluded?: (input: TemplateAgentPolicyInput) => boolean,
   prerelease?: boolean,
-  ignoreUnverified?: boolean
+  ignoreUnverified?: boolean,
+  injectWiredStreamsRouting?: boolean
 ): Promise<{ inputs: TemplateAgentPolicyInput[] }>;
 export async function getTemplateInputs(
   soClient: SavedObjectsClientContract,
@@ -118,7 +120,8 @@ export async function getTemplateInputs(
   format: Format,
   isInputIncluded: (input: TemplateAgentPolicyInput) => boolean = () => true,
   prerelease?: boolean,
-  ignoreUnverified?: boolean
+  ignoreUnverified?: boolean,
+  injectWiredStreamsRouting: boolean = false
 ) {
   const experimentalFeature = appContextService.getExperimentalFeatures();
 
@@ -195,6 +198,28 @@ export async function getTemplateInputs(
     inputIdsDestinationMap
   ).filter(isInputIncluded);
 
+  if (injectWiredStreamsRouting) {
+    for (const input of inputs) {
+      const inputStreams = input.streams as Array<{
+        data_stream?: { type?: string };
+        processors?: Array<Record<string, unknown>>;
+      }>;
+      if (inputStreams) {
+        for (const stream of inputStreams) {
+          if (stream.data_stream?.type === 'logs') {
+            stream.processors = stream.processors || [];
+            stream.processors.unshift({
+              add_fields: {
+                target: '@metadata',
+                fields: { raw_index: 'logs.ecs' },
+              },
+            });
+          }
+        }
+      }
+    }
+  }
+
   let otelcolConfig;
   if (experimentalFeature.enableOtelIntegrations) {
     // Template inputs don't have package info cache, so pass undefined

x-pack/platform/plugins/shared/fleet/server/services/epm/packages/get_templates_inputs.test.ts

@@ -384,4 +384,85 @@ describe('Fleet - getTemplateInputs', () => {
 
     expect(template).toMatchSnapshot();
   });
+
+  it('should inject wired streams routing processor for log streams when enabled', async () => {
+    const soMock = savedObjectsClientMock.create();
+    soMock.get.mockResolvedValue({ attributes: {} } as any);
+    const template = await getTemplateInputs(
+      soMock,
+      'redis',
+      '1.18.0',
+      'json',
+      undefined,
+      undefined,
+      undefined,
+      true // injectWiredStreamsRouting
+    );
+
+    const logInput = template.inputs.find((input) => input.type === 'logfile');
+    expect(logInput).toBeDefined();
+    if (logInput && logInput.streams) {
+      for (const stream of logInput.streams as Array<{
+        data_stream?: { type?: string };
+        processors?: Array<{ add_fields?: { target: string; fields: { raw_index: string } } }>;
+      }>) {
+        if (stream.data_stream?.type === 'logs') {
+          expect(stream.processors).toBeDefined();
+          expect(stream.processors![0]).toEqual({
+            add_fields: {
+              target: '@metadata',
+              fields: { raw_index: 'logs.ecs' },
+            },
+          });
+        }
+      }
+    }
+  });
+
+  it('should NOT inject wired streams routing processor when disabled', async () => {
+    const soMock = savedObjectsClientMock.create();
+    soMock.get.mockResolvedValue({ attributes: {} } as any);
+    const templateWithRouting = await getTemplateInputs(
+      soMock,
+      'redis',
+      '1.18.0',
+      'json',
+      undefined,
+      undefined,
+      undefined,
+      true
+    );
+    const templateWithoutRouting = await getTemplateInputs(
+      soMock,
+      'redis',
+      '1.18.0',
+      'json',
+      undefined,
+      undefined,
+      undefined,
+      false
+    );
+
+    // With routing should have the processor
+    const logInputWithRouting = templateWithRouting.inputs.find(
+      (input) => input.type === 'logfile'
+    );
+    const logInputWithoutRouting = templateWithoutRouting.inputs.find(
+      (input) => input.type === 'logfile'
+    );
+
+    if (logInputWithRouting?.streams && logInputWithoutRouting?.streams) {
+      const streamWithRouting = (
+        logInputWithRouting.streams as Array<{ processors?: unknown[] }>
+      )[0];
+      const streamWithoutRouting = (
+        logInputWithoutRouting.streams as Array<{ processors?: unknown[] }>
+      )[0];
+
+      const routingProcessorCount = streamWithRouting.processors?.length ?? 0;
+      const noRoutingProcessorCount = streamWithoutRouting.processors?.length ?? 0;
+
+      expect(routingProcessorCount).toBe(noRoutingProcessorCount + 1);
+    }
+  });
 });

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

@@ -337,6 +337,9 @@ describe('generateLayer', () => {
       });
       // No attributes passthrough object for ECS streams
       expect(properties.attributes).toBeUndefined();
+
+      // ECS streams use subobjects: false to keep dotted field names as flat keys
+      expect(result.template.mappings?.subobjects).toBe(false);
     });
 
     it('should use OTel settings for OTel-based root streams', () => {
@@ -418,6 +421,9 @@ describe('generateLayer', () => {
       });
       // No attributes passthrough object for ECS streams
       expect(properties.attributes).toBeUndefined();
+
+      // ECS child streams also use subobjects: false
+      expect(result.template.mappings?.subobjects).toBe(false);
     });
   });
 });

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

@@ -221,6 +221,7 @@ export function generateLayer(
       mappings: {
         dynamic: false,
         properties: mappingProperties,
+        ...(isEcsStream && { subobjects: false as const }),
       },
     },
     version: ASSET_VERSION,

x-pack/platform/test/api_integration_deployment_agnostic/apis/streams/basic.ts

@@ -11,6 +11,7 @@ import { Streams, emptyAssets } from '@kbn/streams-schema';
 import { MAX_PRIORITY } from '@kbn/streams-plugin/server/lib/streams/index_templates/generate_index_template';
 import type { InheritedFieldDefinition } from '@kbn/streams-schema/src/fields';
 import { get, omit } from 'lodash';
+import type { JsonObject } from '@kbn/utility-types';
 import type { DeploymentAgnosticFtrProviderContext } from '../../ftr_provider_context';
 import type { StreamsSupertestRepositoryClient } from './helpers/repository_client';
 import {
@@ -473,8 +474,8 @@ export default function ({ getService }: DeploymentAgnosticFtrProviderContext) {
         expect(result._index).to.match(new RegExp(`^\\.ds\\-${rootStream.replace('.', '\\.')}-.*`));
         expect(result._source).to.have.property('@timestamp', '2024-01-01T00:00:00.000Z');
         expect(result._source).to.have.property('message', 'test message');
-        expect(result._source).to.have.property('stream');
-        expect((result._source as any).stream).to.have.property('name', rootStream);
+        // With subobjects: false, stream.name is a flat dotted key
+        expect((result._source as JsonObject)['stream.name']).to.eql(rootStream);
       });
 
       it('Index an ECS doc with nested fields', async () => {
@@ -492,8 +493,7 @@ export default function ({ getService }: DeploymentAgnosticFtrProviderContext) {
         const result = await indexAndAssertTargetStream(esClient, rootStream, doc);
         expect(result._source).to.have.property('@timestamp', '2024-01-01T00:00:00.000Z');
         expect(result._source).to.have.property('message', 'test message');
-        expect(result._source).to.have.property('stream');
-        expect((result._source as any).stream).to.have.property('name', rootStream);
+        expect((result._source as JsonObject)['stream.name']).to.eql(rootStream);
       });
 
       it(`Fork ${rootStream} to ${rootStream}.apache`, async () => {
@@ -541,8 +541,7 @@ export default function ({ getService }: DeploymentAgnosticFtrProviderContext) {
         const result = await indexAndAssertTargetStream(esClient, `${rootStream}.apache`, doc);
         expect(result._source).to.have.property('@timestamp', '2024-01-01T00:00:10.000Z');
         expect(result._source).to.have.property('message', 'Apache access log');
-        expect(result._source).to.have.property('stream');
-        expect((result._source as any).stream).to.have.property('name', `${rootStream}.apache`);
+        expect((result._source as JsonObject)['stream.name']).to.eql(`${rootStream}.apache`);
       });
 
       it(`Fork ${rootStream}.apache to ${rootStream}.apache.error`, async () => {
@@ -574,11 +573,7 @@ export default function ({ getService }: DeploymentAgnosticFtrProviderContext) {
         );
         expect(result._source).to.have.property('@timestamp', '2024-01-01T00:00:20.000Z');
         expect(result._source).to.have.property('message', 'Apache error log');
-        expect(result._source).to.have.property('stream');
-        expect((result._source as any).stream).to.have.property(
-          'name',
-          `${rootStream}.apache.error`
-        );
+        expect((result._source as JsonObject)['stream.name']).to.eql(`${rootStream}.apache.error`);
       });
 
       it(`Does not index to ${rootStream}.apache.error if routing is disabled`, async () => {
@@ -912,11 +907,17 @@ export default function ({ getService }: DeploymentAgnosticFtrProviderContext) {
         }
       }
 
+      const isEcs = streams[0].startsWith('logs.ecs');
       const mappingsResponse = await esClient.indices.getMapping({ index: streams });
       for (const { mappings } of Object.values(mappingsResponse)) {
         for (const [field, fieldConfig] of Object.entries(expectedFields)) {
-          const fieldPath = field.split('.').join('.properties.');
-          expect(get(mappings.properties, fieldPath)).to.eql(omit(fieldConfig, ['from']));
+          if (isEcs) {
+            // With subobjects: false, dotted field names are literal keys in mappings
+            expect(get(mappings.properties, [field])).to.eql(omit(fieldConfig, ['from']));
+          } else {
+            const fieldPath = field.split('.').join('.properties.');
+            expect(get(mappings.properties, fieldPath)).to.eql(omit(fieldConfig, ['from']));
+          }
         }
       }
     }

x-pack/solutions/observability/plugins/observability_onboarding/common/telemetry_events.ts

@@ -337,3 +337,33 @@ export const OBSERVABILITY_ONBOARDING_FLOW_DATASET_DETECTED_TELEMETRY_EVENT: Eve
     context: flowContextSchema,
   },
 };
+
+export const OBSERVABILITY_ONBOARDING_WIRED_STREAMS_AUTO_ENABLED_EVENT: EventTypeOpts<{
+  flow_type: string;
+  success: boolean;
+  error_message?: string;
+}> = {
+  eventType: 'observability_onboarding_wired_streams_auto_enabled',
+  schema: {
+    flow_type: {
+      type: 'keyword',
+      _meta: {
+        description:
+          'The onboarding flow type where auto-enable was triggered (otel_host, otel_kubernetes, elastic_agent_kubernetes, auto_detect)',
+      },
+    },
+    success: {
+      type: 'boolean',
+      _meta: {
+        description: 'Whether the auto-enable operation succeeded',
+      },
+    },
+    error_message: {
+      type: 'text',
+      _meta: {
+        description: 'Error message if auto-enable failed',
+        optional: true,
+      },
+    },
+  },
+};

x-pack/solutions/observability/plugins/observability_onboarding/kibana.jsonc

@@ -27,6 +27,7 @@
     "optionalPlugins": [
       "cloud",
       "usageCollection",
+      "streams",
     ],
     "requiredBundles": [
       "kibanaReact"

x-pack/solutions/observability/plugins/observability_onboarding/moon.yml

@@ -54,6 +54,8 @@ dependsOn:
   - '@kbn/react-kibana-mount'
   - '@kbn/react-query'
   - '@kbn/core-security-server'
+  - '@kbn/streams-plugin'
+  - '@kbn/data-views-plugin'
 tags:
   - plugin
   - prod