[8.19] [Streams 🌊] Enrichment sampling data sources (#219736)

x-pack/platform/plugins/shared/streams/server/routes/internal/streams/management/route.ts

@@ -5,13 +5,7 @@
  * 2.0.
  */
 
-import {
-  SampleDocument,
-  Streams,
-  conditionSchema,
-  conditionToQueryDsl,
-  getConditionFields,
-} from '@kbn/streams-schema';
+import { Streams } from '@kbn/streams-schema';
 import { z } from '@kbn/zod';
 import { STREAMS_API_PRIVILEGES } from '../../../../../common/constants';
 import { SecurityError } from '../../../../lib/streams/errors/security_error';
@@ -23,86 +17,6 @@ import {
 } from '../../../../lib/streams/stream_crud';
 import { createServerRoute } from '../../../create_server_route';
 
-export const sampleStreamRoute = createServerRoute({
-  endpoint: 'POST /internal/streams/{name}/_sample',
-  options: {
-    access: 'internal',
-  },
-  security: {
-    authz: {
-      requiredPrivileges: [STREAMS_API_PRIVILEGES.read],
-    },
-  },
-  params: z.object({
-    path: z.object({ name: z.string() }),
-    body: z.object({
-      if: z.optional(conditionSchema),
-      start: z.optional(z.number()),
-      end: z.optional(z.number()),
-      size: z.optional(z.number()),
-    }),
-  }),
-  handler: async ({ params, request, getScopedClients }) => {
-    const { scopedClusterClient } = await getScopedClients({ request });
-
-    const { read } = await checkAccess({ name: params.path.name, scopedClusterClient });
-
-    if (!read) {
-      throw new SecurityError(`Cannot read stream ${params.path.name}, insufficient privileges`);
-    }
-
-    const { if: condition, start, end, size } = params.body;
-    const searchBody = {
-      query: {
-        bool: {
-          must: [
-            condition ? conditionToQueryDsl(condition) : { match_all: {} },
-            {
-              range: {
-                '@timestamp': {
-                  gte: start,
-                  lte: end,
-                  format: 'epoch_millis',
-                },
-              },
-            },
-          ],
-        },
-      },
-      // Conditions could be using fields which are not indexed or they could use it with other types than they are eventually mapped as.
-      // Because of this we can't rely on mapped fields to draw a sample, instead we need to use runtime fields to simulate what happens during
-      // ingest in the painless condition checks.
-      // This is less efficient than it could be - in some cases, these fields _are_ indexed with the right type and we could use them directly.
-      // This can be optimized in the future.
-      runtime_mappings: condition
-        ? Object.fromEntries(
-            getConditionFields(condition).map((field) => [
-              field.name,
-              { type: field.type === 'string' ? ('keyword' as const) : ('double' as const) },
-            ])
-          )
-        : undefined,
-      sort: [
-        {
-          '@timestamp': {
-            order: 'desc' as const,
-          },
-        },
-      ],
-      terminate_after: size,
-      track_total_hits: false,
-      size,
-    };
-    const results = await scopedClusterClient.asCurrentUser.search({
-      index: params.path.name,
-      allow_no_indices: true,
-      ...searchBody,
-    });
-
-    return { documents: results.hits.hits.map((hit) => hit._source) as SampleDocument[] };
-  },
-});
-
 export const unmanagedAssetDetailsRoute = createServerRoute({
   endpoint: 'GET /internal/streams/{name}/_unmanaged_assets',
   options: {
@@ -148,6 +62,5 @@ export const unmanagedAssetDetailsRoute = createServerRoute({
 });
 
 export const internalManagementRoutes = {
-  ...sampleStreamRoute,
   ...unmanagedAssetDetailsRoute,
 };

x-pack/platform/plugins/shared/streams/server/routes/internal/streams/processing/simulation_handler.ts

@@ -535,9 +535,9 @@ const extractProcessorMetrics = ({
     return {
       detected_fields,
       errors,
-      failed_rate: parseFloat(failureRate.toFixed(2)),
-      skipped_rate: parseFloat(skippedRate.toFixed(2)),
-      parsed_rate: parseFloat(parsedRate.toFixed(2)),
+      failed_rate: parseFloat(failureRate.toFixed(3)),
+      skipped_rate: parseFloat(skippedRate.toFixed(3)),
+      parsed_rate: parseFloat(parsedRate.toFixed(3)),
     };
   });
 };
@@ -718,10 +718,10 @@ const prepareSimulationResponse = async (
     documents: docReports,
     processors_metrics: processorsMetrics,
     documents_metrics: {
-      failed_rate: parseFloat(failureRate.toFixed(2)),
-      partially_parsed_rate: parseFloat(partiallyParsedRate.toFixed(2)),
-      skipped_rate: parseFloat(skippedRate.toFixed(2)),
-      parsed_rate: parseFloat(parsedRate.toFixed(2)),
+      failed_rate: parseFloat(failureRate.toFixed(3)),
+      partially_parsed_rate: parseFloat(partiallyParsedRate.toFixed(3)),
+      skipped_rate: parseFloat(skippedRate.toFixed(3)),
+      parsed_rate: parseFloat(parsedRate.toFixed(3)),
     },
     is_non_additive_simulation: isNotAdditiveSimulation,
   };

x-pack/platform/plugins/shared/streams_app/common/url_schema/common.ts

@@ -0,0 +1,8 @@
+/*
+ * 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 ENRICHMENT_URL_STATE_KEY = 'pageState';

x-pack/platform/plugins/shared/streams_app/common/url_schema/enrichment_url_schema.ts

@@ -0,0 +1,111 @@
+/*
+ * 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 { Filter, TimeRange } from '@kbn/es-query';
+import { SampleDocument, sampleDocument } from '@kbn/streams-schema/src/shared/record_types';
+import { z } from '@kbn/zod';
+
+/**
+ * Base interface for all data source types with common properties
+ */
+export interface BaseDataSource {
+  enabled: boolean;
+  name?: string;
+}
+
+/**
+ * Base schema for common data source properties
+ */
+const baseDataSourceSchema = z.object({
+  enabled: z.boolean(),
+  name: z.string().optional(),
+}) satisfies z.Schema<BaseDataSource>;
+
+/**
+ * Random samples data source that retrieves data from the stream index
+ */
+export interface RandomSamplesDataSource extends BaseDataSource {
+  type: 'random-samples';
+}
+
+const randomSamplesDataSourceSchema = baseDataSourceSchema.extend({
+  type: z.literal('random-samples'),
+}) satisfies z.Schema<RandomSamplesDataSource>;
+
+/**
+ * KQL samples data source that retrieves data based on KQL query
+ */
+export interface KqlSamplesDataSource extends BaseDataSource {
+  type: 'kql-samples';
+  query: {
+    language: string;
+    query: string;
+  };
+  filters?: Filter[];
+  timeRange: TimeRange;
+}
+
+const kqlSamplesDataSourceSchema = baseDataSourceSchema.extend({
+  type: z.literal('kql-samples'),
+  query: z.object({
+    language: z.string(),
+    query: z.string(),
+  }),
+  filters: z.array(z.any()).optional(),
+  timeRange: z.object({
+    from: z.string(),
+    to: z.string(),
+  }),
+}) satisfies z.Schema<KqlSamplesDataSource>;
+
+/**
+ * Custom samples data source with user-provided documents
+ */
+export interface CustomSamplesDataSource extends BaseDataSource {
+  type: 'custom-samples';
+  documents: SampleDocument[];
+}
+
+export const customSamplesDataSourceDocumentsSchema = z.array(sampleDocument);
+
+export const customSamplesDataSourceSchema = baseDataSourceSchema.extend({
+  type: z.literal('custom-samples'),
+  documents: customSamplesDataSourceDocumentsSchema,
+}) satisfies z.Schema<CustomSamplesDataSource>;
+
+/**
+ * Union type of all possible data source types
+ */
+export type EnrichmentDataSource =
+  | RandomSamplesDataSource
+  | KqlSamplesDataSource
+  | CustomSamplesDataSource;
+
+/**
+ * Schema for validating enrichment data sources
+ */
+const enrichmentDataSourceSchema = z.union([
+  randomSamplesDataSourceSchema,
+  kqlSamplesDataSourceSchema,
+  customSamplesDataSourceSchema,
+]) satisfies z.Schema<EnrichmentDataSource>;
+
+/**
+ * URL state for enrichment configuration
+ */
+export interface EnrichmentUrlState {
+  v: 1;
+  dataSources: EnrichmentDataSource[];
+}
+
+/**
+ * Schema for validating enrichment URL state
+ */
+export const enrichmentUrlSchema = z.object({
+  v: z.literal(1),
+  dataSources: z.array(enrichmentDataSourceSchema),
+}) satisfies z.Schema<EnrichmentUrlState>;

x-pack/platform/plugins/shared/streams_app/common/url_schema/index.ts

@@ -0,0 +1,9 @@
+/*
+ * 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 { ENRICHMENT_URL_STATE_KEY } from './common';
+export * from './enrichment_url_schema';

x-pack/platform/plugins/shared/streams_app/kibana.jsonc

@@ -28,6 +28,6 @@
       "streams",
       "unifiedSearch"
     ],
-    "requiredBundles": ["kibanaReact"]
+    "requiredBundles": ["kibanaReact", "kibanaUtils"]
   }
 }

x-pack/platform/plugins/shared/streams_app/public/components/app_root/index.tsx

@@ -17,6 +17,7 @@ import { StreamsAppContextProvider } from '../streams_app_context_provider';
 import { streamsAppRouter } from '../../routes/config';
 import { StreamsAppStartDependencies } from '../../types';
 import { StreamsAppServices } from '../../services/types';
+import { KbnUrlStateStorageFromRouterProvider } from '../../util/kbn_url_state_context';
 
 export function AppRoot({
   coreStart,
@@ -46,9 +47,11 @@ export function AppRoot({
     <StreamsAppContextProvider context={context}>
       <RedirectAppLinks coreStart={coreStart}>
         <RouterProvider history={history} router={streamsAppRouter}>
-          <BreadcrumbsContextProvider>
-            <RouteRenderer />
-          </BreadcrumbsContextProvider>
+          <KbnUrlStateStorageFromRouterProvider>
+            <BreadcrumbsContextProvider>
+              <RouteRenderer />
+            </BreadcrumbsContextProvider>
+          </KbnUrlStateStorageFromRouterProvider>
         </RouterProvider>
       </RedirectAppLinks>
     </StreamsAppContextProvider>

x-pack/platform/plugins/shared/streams_app/public/components/data_management/preview_table/index.tsx

@@ -4,14 +4,15 @@
  * 2.0; you may not use this file except in compliance with the Elastic License
  * 2.0.
  */
-import { EuiDataGrid, EuiDataGridRowHeightsOptions } from '@elastic/eui';
+import { EuiDataGrid, EuiDataGridProps, EuiDataGridRowHeightsOptions } from '@elastic/eui';
 import { i18n } from '@kbn/i18n';
 import { SampleDocument } from '@kbn/streams-schema';
 import React, { useMemo } from 'react';
 
 export function PreviewTable({
   documents,
   displayColumns,
+  height,
   renderCellValue,
   rowHeightsOptions,
   toolbarVisibility = false,
@@ -20,6 +21,7 @@ export function PreviewTable({
 }: {
   documents: SampleDocument[];
   displayColumns?: string[];
+  height?: EuiDataGridProps['height'];
   renderCellValue?: (doc: SampleDocument, columnId: string) => React.ReactNode | undefined;
   rowHeightsOptions?: EuiDataGridRowHeightsOptions;
   toolbarVisibility?: boolean;
@@ -97,6 +99,7 @@ export function PreviewTable({
         setVisibleColumns: setVisibleColumns || (() => {}),
         canDragAndDropColumns: false,
       }}
+      height={height}
       toolbarVisibility={toolbarVisibility}
       rowCount={documents.length}
       rowHeightsOptions={rowHeightsOptions}

x-pack/platform/plugins/shared/streams_app/public/components/data_management/stream_detail_enrichment/data_sources_flyout/add_data_sources_context_menu.tsx

@@ -0,0 +1,64 @@
+/*
+ * 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, EuiContextMenu, EuiPopover } from '@elastic/eui';
+import { useBoolean } from '@kbn/react-hooks';
+import { DATA_SOURCES_I18N } from './translations';
+import {
+  defaultCustomSamplesDataSource,
+  defaultKqlSamplesDataSource,
+} from '../state_management/stream_enrichment_state_machine/utils';
+import { useStreamEnrichmentEvents } from '../state_management/stream_enrichment_state_machine';
+
+export const AddDataSourcesContextMenu = () => {
+  const { addDataSource } = useStreamEnrichmentEvents();
+
+  const [isOpen, { toggle: toggleMenu, off: closeMenu }] = useBoolean();
+
+  return (
+    <EuiPopover
+      id="data-sources-menu"
+      button={
+        <EuiButton size="s" iconType="arrowDown" iconSide="right" onClick={toggleMenu}>
+          {DATA_SOURCES_I18N.contextMenu.addDataSource}
+        </EuiButton>
+      }
+      isOpen={isOpen}
+      closePopover={closeMenu}
+      panelPaddingSize="none"
+      anchorPosition="downLeft"
+    >
+      <EuiContextMenu
+        initialPanelId="data-source-options"
+        panels={[
+          {
+            id: 'data-source-options',
+            items: [
+              {
+                name: DATA_SOURCES_I18N.contextMenu.addKqlDataSource,
+                icon: 'search',
+                onClick: () => {
+                  addDataSource(defaultKqlSamplesDataSource);
+                  closeMenu();
+                },
+              },
+              {
+                name: DATA_SOURCES_I18N.contextMenu.addCustomSamples,
+                icon: 'visText',
+                onClick: () => {
+                  addDataSource(defaultCustomSamplesDataSource);
+                  closeMenu();
+                },
+              },
+            ],
+          },
+        ]}
+      />
+    </EuiPopover>
+  );
+};