🌊 Streams routing UI

src/plugins/unified_search/public/query_string_input/query_bar_top_row.tsx

@@ -787,12 +787,17 @@ export const QueryBarTopRow = React.memo(
                   adHocDataview={props.indexPatterns?.[0]}
                 />
               )}
-              <EuiFlexItem
-                grow={!shouldShowDatePickerAsBadge()}
-                style={{ minWidth: shouldShowDatePickerAsBadge() ? 'auto' : 320, maxWidth: '100%' }}
-              >
-                {!isQueryLangSelected ? renderQueryInput() : null}
-              </EuiFlexItem>
+              {!isQueryLangSelected && shouldRenderQueryInput() && (
+                <EuiFlexItem
+                  grow={!shouldShowDatePickerAsBadge()}
+                  style={{
+                    minWidth: shouldShowDatePickerAsBadge() ? 'auto' : 320,
+                    maxWidth: '100%',
+                  }}
+                >
+                  {renderQueryInput()}
+                </EuiFlexItem>
+              )}
               {props.renderQueryInputAppend?.()}
               {shouldShowDatePickerAsBadge() && props.filterBar}
               {renderUpdateButton()}

x-pack/plugins/streams/common/index.ts

@@ -5,4 +5,4 @@
  * 2.0.
  */
 
-export type { StreamDefinition } from './types';
+export type { StreamDefinition, ReadStreamDefinition } from './types';

x-pack/plugins/streams/common/types.ts

@@ -33,11 +33,11 @@ export interface AndCondition {
   and: Condition[];
 }
 
-export interface RerouteOrCondition {
+export interface OrCondition {
   or: Condition[];
 }
 
-export type Condition = FilterCondition | AndCondition | RerouteOrCondition | undefined;
+export type Condition = FilterCondition | AndCondition | OrCondition | undefined;
 
 export const conditionSchema: z.ZodType<Condition> = z.lazy(() =>
   z.union([
@@ -77,17 +77,17 @@ export const fieldDefinitionSchema = z.object({
 
 export type FieldDefinition = z.infer<typeof fieldDefinitionSchema>;
 
+export const streamChildSchema = z.object({
+  id: z.string(),
+  condition: z.optional(conditionSchema),
+});
+
+export type StreamChild = z.infer<typeof streamChildSchema>;
+
 export const streamWithoutIdDefinitonSchema = z.object({
   processing: z.array(processingDefinitionSchema).default([]),
   fields: z.array(fieldDefinitionSchema).default([]),
-  children: z
-    .array(
-      z.object({
-        id: z.string(),
-        condition: z.optional(conditionSchema),
-      })
-    )
-    .default([]),
+  children: z.array(streamChildSchema).default([]),
 });
 
 export type StreamWithoutIdDefinition = z.infer<typeof streamDefinitonSchema>;
@@ -101,3 +101,9 @@ export type StreamDefinition = z.infer<typeof streamDefinitonSchema>;
 export const streamDefinitonWithoutChildrenSchema = streamDefinitonSchema.omit({ children: true });
 
 export type StreamWithoutChildrenDefinition = z.infer<typeof streamDefinitonWithoutChildrenSchema>;
+
+export const readStreamDefinitonSchema = streamDefinitonSchema.extend({
+  inheritedFields: z.array(fieldDefinitionSchema.extend({ from: z.string() })).default([]),
+});
+
+export type ReadStreamDefinition = z.infer<typeof readStreamDefinitonSchema>;

x-pack/plugins/streams/server/lib/streams/helpers/condition_fields.ts

@@ -0,0 +1,85 @@
+/*
+ * 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 { Condition, FilterCondition } from '../../../../common/types';
+import { isAndCondition, isFilterCondition, isOrCondition } from './condition_guards';
+
+export function isComplete(condition: Condition): boolean {
+  if (isFilterCondition(condition)) {
+    return condition.field !== undefined && condition.field !== '';
+  }
+  if (isAndCondition(condition)) {
+    return condition.and.every(isComplete);
+  }
+  if (isOrCondition(condition)) {
+    return condition.or.every(isComplete);
+  }
+  return false;
+}
+
+export function getFields(
+  condition: Condition
+): Array<{ name: string; type: 'number' | 'string' }> {
+  const fields = collectFields(condition);
+  // deduplicate fields, if mapped as string and number, keep as number
+  const uniqueFields = new Map<string, 'number' | 'string'>();
+  fields.forEach((field) => {
+    const existing = uniqueFields.get(field.name);
+    if (existing === 'number') {
+      return;
+    }
+    if (existing === 'string' && field.type === 'number') {
+      uniqueFields.set(field.name, 'number');
+      return;
+    }
+    uniqueFields.set(field.name, field.type);
+  });
+
+  return Array.from(uniqueFields).map(([name, type]) => ({ name, type }));
+}
+
+function collectFields(condition: Condition): Array<{ name: string; type: 'number' | 'string' }> {
+  if (isFilterCondition(condition)) {
+    return [{ name: condition.field, type: getFieldTypeForFilterCondition(condition) }];
+  }
+  if (isAndCondition(condition)) {
+    return condition.and.flatMap(collectFields);
+  }
+  if (isOrCondition(condition)) {
+    return condition.or.flatMap(collectFields);
+  }
+  return [];
+}
+
+function getFieldTypeForFilterCondition(condition: FilterCondition): 'number' | 'string' {
+  switch (condition.operator) {
+    case 'gt':
+    case 'gte':
+    case 'lt':
+    case 'lte':
+      return 'number';
+    case 'neq':
+    case 'eq':
+    case 'exists':
+    case 'contains':
+    case 'startsWith':
+    case 'endsWith':
+    case 'notExists':
+      return 'string';
+    default:
+      return 'string';
+  }
+}
+
+export function validateCondition(condition: Condition) {
+  if (isFilterCondition(condition)) {
+    // check whether a field is specified
+    if (!condition.field.trim()) {
+      throw new Error('Field is required in conditions');
+    }
+  }
+}

x-pack/plugins/streams/server/lib/streams/helpers/condition_guards.ts

@@ -0,0 +1,29 @@
+/*
+ * 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 {
+  AndCondition,
+  conditionSchema,
+  FilterCondition,
+  filterConditionSchema,
+  OrCondition,
+} from '../../../../common/types';
+
+export function isFilterCondition(subject: any): subject is FilterCondition {
+  const result = filterConditionSchema.safeParse(subject);
+  return result.success;
+}
+
+export function isAndCondition(subject: any): subject is AndCondition {
+  const result = conditionSchema.safeParse(subject);
+  return result.success && subject.and != null;
+}
+
+export function isOrCondition(subject: any): subject is OrCondition {
+  const result = conditionSchema.safeParse(subject);
+  return result.success && subject.or != null;
+}

x-pack/plugins/streams/server/lib/streams/helpers/condition_to_painless.ts

@@ -7,30 +7,12 @@
 
 import { isBoolean, isString } from 'lodash';
 import {
-  AndCondition,
   BinaryFilterCondition,
   Condition,
-  conditionSchema,
   FilterCondition,
-  filterConditionSchema,
-  RerouteOrCondition,
   UnaryFilterCondition,
 } from '../../../../common/types';
-
-function isFilterCondition(subject: any): subject is FilterCondition {
-  const result = filterConditionSchema.safeParse(subject);
-  return result.success;
-}
-
-function isAndCondition(subject: any): subject is AndCondition {
-  const result = conditionSchema.safeParse(subject);
-  return result.success && subject.and != null;
-}
-
-function isOrCondition(subject: any): subject is RerouteOrCondition {
-  const result = conditionSchema.safeParse(subject);
-  return result.success && subject.or != null;
-}
+import { isAndCondition, isFilterCondition, isOrCondition } from './condition_guards';
 
 function safePainlessField(condition: FilterCondition) {
   return `ctx.${condition.field.split('.').join('?.')}`;

x-pack/plugins/streams/server/lib/streams/helpers/condition_to_query_dsl.ts

@@ -0,0 +1,63 @@
+/*
+ * 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 { Condition, FilterCondition } from '../../../../common/types';
+import { isAndCondition, isFilterCondition, isOrCondition } from './condition_guards';
+
+function conditionToClause(condition: FilterCondition) {
+  switch (condition.operator) {
+    case 'neq':
+      return { bool: { must_not: { match: { [condition.field]: condition.value } } } };
+    case 'eq':
+      return { match: { [condition.field]: condition.value } };
+    case 'exists':
+      return { exists: { field: condition.field } };
+    case 'gt':
+      return { range: { [condition.field]: { gt: condition.value } } };
+    case 'gte':
+      return { range: { [condition.field]: { gte: condition.value } } };
+    case 'lt':
+      return { range: { [condition.field]: { lt: condition.value } } };
+    case 'lte':
+      return { range: { [condition.field]: { lte: condition.value } } };
+    case 'contains':
+      return { wildcard: { [condition.field]: `*${condition.value}*` } };
+    case 'startsWith':
+      return { prefix: { [condition.field]: condition.value } };
+    case 'endsWith':
+      return { wildcard: { [condition.field]: `*${condition.value}` } };
+    case 'notExists':
+      return { bool: { must_not: { exists: { field: condition.field } } } };
+    default:
+      return { match_none: {} };
+  }
+}
+
+export function conditionToQueryDsl(condition: Condition): any {
+  if (isFilterCondition(condition)) {
+    return conditionToClause(condition);
+  }
+  if (isAndCondition(condition)) {
+    const and = condition.and.map((filter) => conditionToQueryDsl(filter));
+    return {
+      bool: {
+        must: and,
+      },
+    };
+  }
+  if (isOrCondition(condition)) {
+    const or = condition.or.map((filter) => conditionToQueryDsl(filter));
+    return {
+      bool: {
+        should: or,
+      },
+    };
+  }
+  return {
+    match_none: {},
+  };
+}

x-pack/plugins/streams/server/lib/streams/ingest_pipelines/logs_default_pipeline.ts

@@ -20,4 +20,9 @@ export const logsDefaultPipelineProcessors = [
       ignore_missing_pipeline: true,
     },
   },
+  {
+    dot_expander: {
+      field: '*',
+    },
+  },
 ];

x-pack/plugins/streams/server/lib/streams/stream_crud.ts

@@ -102,11 +102,12 @@ export async function listStreams({
     sort: [{ id: 'asc' }],
   });
   const definitions = response.hits.hits.map((hit) => hit._source!);
-  const total = response.hits.total!;
-
+  const hasAccess = await Promise.all(
+    definitions.map((definition) => checkReadAccess({ id: definition.id, scopedClusterClient }))
+  );
   return {
-    definitions,
-    total: typeof total === 'number' ? total : total.value,
+    definitions: definitions.filter((_, index) => hasAccess[index]),
+    total: definitions.length,
   };
 }
 
@@ -163,7 +164,9 @@ export async function readAncestors({
 
   return {
     ancestors: await Promise.all(
-      ancestorIds.map((ancestorId) => readStream({ scopedClusterClient, id: ancestorId }))
+      ancestorIds.map((ancestorId) =>
+        readStream({ scopedClusterClient, id: ancestorId, skipAccessCheck: true })
+      )
     ),
   };
 }

x-pack/plugins/streams/server/routes/index.ts

@@ -14,6 +14,7 @@ import { forkStreamsRoute } from './streams/fork';
 import { listStreamsRoute } from './streams/list';
 import { readStreamRoute } from './streams/read';
 import { resyncStreamsRoute } from './streams/resync';
+import { sampleStreamRoute } from './streams/sample';
 import { streamsStatusRoutes } from './streams/settings';
 
 export const streamsRouteRepository = {
@@ -27,6 +28,7 @@ export const streamsRouteRepository = {
   ...streamsStatusRoutes,
   ...esqlRoutes,
   ...disableStreamsRoute,
+  ...sampleStreamRoute,
 };
 
 export type StreamsRouteRepository = typeof streamsRouteRepository;

x-pack/plugins/streams/server/routes/streams/edit.ts

@@ -27,6 +27,7 @@ import {
 import { MalformedStreamId } from '../../lib/streams/errors/malformed_stream_id';
 import { getParentId } from '../../lib/streams/helpers/hierarchy';
 import { MalformedChildren } from '../../lib/streams/errors/malformed_children';
+import { validateCondition } from '../../lib/streams/helpers/condition_fields';
 
 export const editStreamRoute = createServerRoute({
   endpoint: 'PUT /api/streams/{id}',
@@ -57,7 +58,7 @@ export const editStreamRoute = createServerRoute({
       const parentId = getParentId(params.path.id);
       let parentDefinition: StreamDefinition | undefined;
 
-      const streamDefinition = { ...params.body };
+      const streamDefinition = { ...params.body, id: params.path.id };
 
       // always need to go from the leaves to the parent when syncing ingest pipelines, otherwise data
       // will be routed before the data stream is ready
@@ -87,7 +88,7 @@ export const editStreamRoute = createServerRoute({
 
       await syncStream({
         scopedClusterClient,
-        definition: { ...streamDefinition, id: params.path.id },
+        definition: streamDefinition,
         rootDefinition: parentDefinition,
         logger,
       });
@@ -150,7 +151,7 @@ async function updateParentStream(
 async function validateStreamChildren(
   scopedClusterClient: IScopedClusterClient,
   id: string,
-  children: Array<{ id: string }>
+  children: StreamDefinition['children']
 ) {
   try {
     const { definition: oldDefinition } = await readStream({
@@ -159,6 +160,9 @@ async function validateStreamChildren(
     });
     const oldChildren = oldDefinition.children.map((child) => child.id);
     const newChildren = new Set(children.map((child) => child.id));
+    children.forEach((child) => {
+      validateCondition(child.condition);
+    });
     if (oldChildren.some((child) => !newChildren.has(child))) {
       throw new MalformedChildren(
         'Cannot remove children from a stream, please delete the stream instead'