elastic
diff --git a/‎x-pack/solutions/observability/plugins/streams/common/types.ts‎
Lines changed: 1 addition & 0 deletions b/‎x-pack/solutions/observability/plugins/streams/common/types.ts‎
Lines changed: 1 addition & 0 deletions
diff --git a/‎x-pack/solutions/observability/plugins/streams/server/lib/streams/component_templates/generate_layer.ts‎
Lines changed: 3 additions & 0 deletions b/‎x-pack/solutions/observability/plugins/streams/server/lib/streams/component_templates/generate_layer.ts‎
Lines changed: 3 additions & 0 deletions
diff --git a/‎x-pack/solutions/observability/plugins/streams/server/routes/index.ts‎
Lines changed: 4 additions & 0 deletions b/‎x-pack/solutions/observability/plugins/streams/server/routes/index.ts‎
Lines changed: 4 additions & 0 deletions
diff --git a/‎x-pack/solutions/observability/plugins/streams/server/routes/streams/schema/fields_simulation.ts‎
Lines changed: 207 additions & 0 deletions b/‎x-pack/solutions/observability/plugins/streams/server/routes/streams/schema/fields_simulation.ts‎
Lines changed: 207 additions & 0 deletions
diff --git a/‎x-pack/solutions/observability/plugins/streams/server/routes/streams/schema/unmapped_fields.ts‎
Lines changed: 103 additions & 0 deletions b/‎x-pack/solutions/observability/plugins/streams/server/routes/streams/schema/unmapped_fields.ts‎
Lines changed: 103 additions & 0 deletions
diff --git a/‎x-pack/solutions/observability/plugins/streams/tsconfig.json‎
Lines changed: 1 addition & 0 deletions b/‎x-pack/solutions/observability/plugins/streams/tsconfig.json‎
Lines changed: 1 addition & 0 deletions
diff --git a/‎x-pack/solutions/observability/plugins/streams_app/public/components/preview_table/index.tsx‎
Lines changed: 10 additions & 2 deletions b/‎x-pack/solutions/observability/plugins/streams_app/public/components/preview_table/index.tsx‎
Lines changed: 10 additions & 2 deletions
@@ -73,6 +73,7 @@ export type ProcessingDefinition = z.infer<typeof processingDefinitionSchema>;
 export const fieldDefinitionSchema = z.object({
   name: z.string(),
   type: z.enum(['keyword', 'match_only_text', 'long', 'double', 'date', 'boolean', 'ip']),
+  format: z.optional(z.string()),
 });
 
 export type FieldDefinition = z.infer<typeof fieldDefinitionSchema>;
 
@@ -29,6 +29,9 @@ export function generateLayer(
       // @timestamp can't ignore malformed dates as it's used for sorting in logsdb
       (property as MappingDateProperty).ignore_malformed = false;
     }
+    if (field.type === 'date' && field.format) {
+      (property as MappingDateProperty).format = field.format;
+    }
     properties[field.name] = property;
   });
   return {
 
@@ -15,6 +15,8 @@ import { listStreamsRoute } from './streams/list';
 import { readStreamRoute } from './streams/read';
 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 { streamsStatusRoutes } from './streams/settings';
 
 export const streamsRouteRepository = {
@@ -29,6 +31,8 @@ export const streamsRouteRepository = {
   ...esqlRoutes,
   ...disableStreamsRoute,
   ...sampleStreamRoute,
+  ...unmappedFieldsRoute,
+  ...schemaFieldsSimulationRoute,
 };
 
 export type StreamsRouteRepository = typeof streamsRouteRepository;
@@ -0,0 +1,207 @@
+/*
+ * 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 } from '@hapi/boom';
+import { getFlattenedObject } from '@kbn/std';
+import { fieldDefinitionSchema } from '../../../../common/types';
+import { createServerRoute } from '../../create_server_route';
+import { DefinitionNotFound } from '../../../lib/streams/errors';
+import { checkReadAccess } from '../../../lib/streams/stream_crud';
+
+const SAMPLE_SIZE = 200;
+
+export const schemaFieldsSimulationRoute = createServerRoute({
+  endpoint: 'POST /api/streams/{id}/schema/fields_simulation',
+  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({
+      field_definitions: z.array(fieldDefinitionSchema),
+    }),
+  }),
+  handler: async ({
+    response,
+    params,
+    request,
+    logger,
+    getScopedClients,
+  }): Promise<{
+    status: 'unknown' | 'success' | 'failure';
+    simulationError: string | null;
+    documentsWithRuntimeFieldsApplied: unknown[] | null;
+  }> => {
+    try {
+      const { scopedClusterClient } = await getScopedClients({ request });
+
+      const hasAccess = await checkReadAccess({ id: params.path.id, scopedClusterClient });
+      if (!hasAccess) {
+        throw new DefinitionNotFound(`Stream definition for ${params.path.id} not found.`);
+      }
+
+      const propertiesForSample = Object.fromEntries(
+        params.body.field_definitions.map((field) => [field.name, { type: 'keyword' }])
+      );
+
+      const documentSamplesSearchBody = {
+        // Add keyword runtime mappings so we can pair with exists, this is to attempt to "miss" less documents for the simulation.
+        runtime_mappings: propertiesForSample,
+        query: {
+          bool: {
+            filter: Object.keys(propertiesForSample).map((field) => ({
+              exists: { field },
+            })),
+          },
+        },
+        sort: [
+          {
+            '@timestamp': {
+              order: 'desc',
+            },
+          },
+        ],
+        size: SAMPLE_SIZE,
+      };
+
+      const sampleResults = await scopedClusterClient.asCurrentUser.search({
+        index: params.path.id,
+        ...documentSamplesSearchBody,
+      });
+
+      if (
+        (typeof sampleResults.hits.total === 'object' && sampleResults.hits.total?.value === 0) ||
+        sampleResults.hits.total === 0 ||
+        !sampleResults.hits.total
+      ) {
+        return {
+          status: 'unknown',
+          simulationError: null,
+          documentsWithRuntimeFieldsApplied: null,
+        };
+      }
+
+      const propertiesForSimulation = Object.fromEntries(
+        params.body.field_definitions.map((field) => [
+          field.name,
+          { type: field.type, ...(field.format ? { format: field.format } : {}) },
+        ])
+      );
+
+      const fieldDefinitionKeys = Object.keys(propertiesForSimulation);
+
+      const sampleResultsAsSimulationDocs = sampleResults.hits.hits.map((hit) => ({
+        _index: params.path.id,
+        _id: hit._id,
+        _source: Object.fromEntries(
+          Object.entries(getFlattenedObject(hit._source as Record<string, unknown>)).filter(
+            ([k]) => fieldDefinitionKeys.includes(k) || k === '@timestamp'
+          )
+        ),
+      }));
+
+      const simulationBody = {
+        docs: sampleResultsAsSimulationDocs,
+        component_template_substitutions: {
+          [`${params.path.id}@stream.layer`]: {
+            template: {
+              mappings: {
+                dynamic: 'strict',
+                properties: propertiesForSimulation,
+              },
+            },
+          },
+        },
+      };
+
+      // TODO: We should be using scopedClusterClient.asCurrentUser.simulate.ingest() but the ES JS lib currently has a bug. The types also aren't available yet, so we use any.
+      const simulation = (await scopedClusterClient.asCurrentUser.transport.request({
+        method: 'POST',
+        path: `_ingest/_simulate`,
+        body: simulationBody,
+      })) as any;
+
+      const hasErrors = simulation.docs.some((doc: any) => doc.doc.error !== undefined);
+
+      if (hasErrors) {
+        const documentWithError = simulation.docs.find((doc: any) => {
+          return doc.doc.error !== undefined;
+        });
+
+        return {
+          status: 'failure',
+          simulationError: JSON.stringify(
+            // Use the first error as a representative error
+            documentWithError.doc.error
+          ),
+          documentsWithRuntimeFieldsApplied: null,
+        };
+      }
+
+      // Convert the field definitions to a format that can be used in runtime mappings (match_only_text -> keyword)
+      const propertiesCompatibleWithRuntimeMappings = Object.fromEntries(
+        params.body.field_definitions.map((field) => [
+          field.name,
+          {
+            type: field.type === 'match_only_text' ? 'keyword' : field.type,
+            ...(field.format ? { format: field.format } : {}),
+          },
+        ])
+      );
+
+      const runtimeFieldsSearchBody = {
+        runtime_mappings: propertiesCompatibleWithRuntimeMappings,
+        sort: [
+          {
+            '@timestamp': {
+              order: 'desc',
+            },
+          },
+        ],
+        size: SAMPLE_SIZE,
+        fields: params.body.field_definitions.map((field) => field.name),
+        _source: false,
+      };
+
+      // This gives us a "fields" representation rather than _source from the simulation
+      const runtimeFieldsResult = await scopedClusterClient.asCurrentUser.search({
+        index: params.path.id,
+        ...runtimeFieldsSearchBody,
+      });
+
+      return {
+        status: 'success',
+        simulationError: null,
+        documentsWithRuntimeFieldsApplied: runtimeFieldsResult.hits.hits
+          .map((hit) => {
+            if (!hit.fields) {
+              return {};
+            }
+            return Object.keys(hit.fields).reduce<Record<string, unknown>>((acc, field) => {
+              acc[field] = hit.fields![field][0];
+              return acc;
+            }, {});
+          })
+          .filter((doc) => Object.keys(doc).length > 0),
+      };
+    } catch (e) {
+      if (e instanceof DefinitionNotFound) {
+        throw notFound(e);
+      }
+
+      throw internal(e);
+    }
+  },
+});
@@ -0,0 +1,103 @@
+/*
+ * 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 { internal, notFound } from '@hapi/boom';
+import { getFlattenedObject } from '@kbn/std';
+import { DefinitionNotFound } from '../../../lib/streams/errors';
+import { checkReadAccess, readAncestors, readStream } from '../../../lib/streams/stream_crud';
+import { createServerRoute } from '../../create_server_route';
+
+const SAMPLE_SIZE = 500;
+
+export const unmappedFieldsRoute = createServerRoute({
+  endpoint: 'GET /api/streams/{id}/schema/unmapped_fields',
+  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() }),
+  }),
+  handler: async ({
+    response,
+    params,
+    request,
+    logger,
+    getScopedClients,
+  }): Promise<{ unmappedFields: string[] }> => {
+    try {
+      const { scopedClusterClient } = await getScopedClients({ request });
+
+      const hasAccess = await checkReadAccess({ id: params.path.id, scopedClusterClient });
+      if (!hasAccess) {
+        throw new DefinitionNotFound(`Stream definition for ${params.path.id} not found.`);
+      }
+
+      const streamEntity = await readStream({
+        scopedClusterClient,
+        id: params.path.id,
+      });
+
+      const searchBody = {
+        sort: [
+          {
+            '@timestamp': {
+              order: 'desc',
+            },
+          },
+        ],
+        size: SAMPLE_SIZE,
+      };
+
+      const results = await scopedClusterClient.asCurrentUser.search({
+        index: params.path.id,
+        ...searchBody,
+      });
+
+      const sourceFields = new Set<string>();
+
+      results.hits.hits.forEach((hit) => {
+        Object.keys(getFlattenedObject(hit._source as Record<string, unknown>)).forEach((field) => {
+          sourceFields.add(field);
+        });
+      });
+
+      // Mapped fields from the stream's definition and inherited from ancestors
+      const mappedFields = new Set<string>();
+
+      streamEntity.definition.fields.forEach((field) => mappedFields.add(field.name));
+
+      const { ancestors } = await readAncestors({
+        id: params.path.id,
+        scopedClusterClient,
+      });
+
+      for (const ancestor of ancestors) {
+        ancestor.definition.fields.forEach((field) => mappedFields.add(field.name));
+      }
+
+      const unmappedFields = Array.from(sourceFields)
+        .filter((field) => !mappedFields.has(field))
+        .sort();
+
+      return { unmappedFields };
+    } catch (e) {
+      if (e instanceof DefinitionNotFound) {
+        throw notFound(e);
+      }
+
+      throw internal(e);
+    }
+  },
+});
@@ -30,6 +30,7 @@
     "@kbn/server-route-repository-client",
     "@kbn/observability-utils-server",
     "@kbn/observability-utils-common",
+    "@kbn/std",
     "@kbn/safer-lodash-set"
   ]
 }
@@ -8,7 +8,13 @@ import { EuiDataGrid } from '@elastic/eui';
 import { i18n } from '@kbn/i18n';
 import React, { useEffect, useMemo, useState } from 'react';
 
-export function PreviewTable({ documents }: { documents: unknown[] }) {
+export function PreviewTable({
+  documents,
+  displayColumns,
+}: {
+  documents: unknown[];
+  displayColumns?: string[];
+}) {
   const [height, setHeight] = useState('100px');
   useEffect(() => {
     // set height to 100% after a short delay otherwise it doesn't calculate correctly
@@ -19,6 +25,8 @@ export function PreviewTable({ documents }: { documents: unknown[] }) {
   }, []);
 
   const columns = useMemo(() => {
+    if (displayColumns) return displayColumns;
+
     const cols = new Set<string>();
     documents.forEach((doc) => {
       if (!doc || typeof doc !== 'object') {
@@ -29,7 +37,7 @@ export function PreviewTable({ documents }: { documents: unknown[] }) {
       });
     });
     return Array.from(cols);
-  }, [documents]);
+  }, [displayColumns, documents]);
 
   const gridColumns = useMemo(() => {
     return Array.from(columns).map((column) => ({
Original file line number	Diff line number	Diff line change
`@@ -30,6 +30,7 @@`
`30`	`30`	`"@kbn/server-route-repository-client",`
`31`	`31`	`"@kbn/observability-utils-server",`
`32`	`32`	`"@kbn/observability-utils-common",`
	`33`	`+ "@kbn/std",`
`33`	`34`	`"@kbn/safer-lodash-set"`
`34`	`35`	`]`
`35`	`36`	`}`