🌊 Add failure store and exists conditions (elastic#200861)

This PR enables failure store for all streams-managed data streams and
also adds exists and notExists operators to the painless condition
builder.

With this addition I think we have all the important ones covered, we
can discuss whether we should add a "matches regex" operator, but I'm
not sure whether it is a good idea to do so.

This also fixes a problem with rollovers - in data streams `@timestamp`
is not allowed to use `ignore_malformed`, but it's assuming it
automatically, so a rollover is always triggered. This is fixed by
setting `ignore_malformed` to false explicitly for the timestamp.

---------

Co-authored-by: Chris Cowan <[email protected]>
Co-authored-by: kibanamachine <[email protected]>

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

@@ -9,13 +9,25 @@ import { z } from '@kbn/zod';
 
 const stringOrNumberOrBoolean = z.union([z.string(), z.number(), z.boolean()]);
 
-export const filterConditionSchema = z.object({
+export const binaryConditionSchema = z.object({
   field: z.string(),
   operator: z.enum(['eq', 'neq', 'lt', 'lte', 'gt', 'gte', 'contains', 'startsWith', 'endsWith']),
   value: stringOrNumberOrBoolean,
 });
 
+export const unaryFilterConditionSchema = z.object({
+  field: z.string(),
+  operator: z.enum(['exists', 'notExists']),
+});
+
+export const filterConditionSchema = z.discriminatedUnion('operator', [
+  unaryFilterConditionSchema,
+  binaryConditionSchema,
+]);
+
 export type FilterCondition = z.infer<typeof filterConditionSchema>;
+export type BinaryFilterCondition = z.infer<typeof binaryConditionSchema>;
+export type UnaryFilterCondition = z.infer<typeof unaryFilterConditionSchema>;
 
 export interface AndCondition {
   and: Condition[];

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

@@ -7,6 +7,7 @@
 
 import {
   ClusterPutComponentTemplateRequest,
+  MappingDateProperty,
   MappingProperty,
 } from '@elastic/elasticsearch/lib/api/types';
 import { StreamDefinition } from '../../../../common/types';
@@ -21,9 +22,14 @@ export function generateLayer(
 ): ClusterPutComponentTemplateRequest {
   const properties: Record<string, MappingProperty> = {};
   definition.fields.forEach((field) => {
-    properties[field.name] = {
+    const property: MappingProperty = {
       type: field.type,
     };
+    if (field.name === '@timestamp') {
+      // @timestamp can't ignore malformed dates as it's used for sorting in logsdb
+      (property as MappingDateProperty).ignore_malformed = false;
+    }
+    properties[field.name] = property;
   });
   return {
     name: getComponentTemplateName(id),

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

@@ -44,6 +44,14 @@ const operatorConditionAndResults = [
     condition: { field: 'log.logger', operator: 'contains' as const, value: 'proxy' },
     result: '(ctx.log?.logger !== null && ctx.log?.logger.contains("proxy"))',
   },
+  {
+    condition: { field: 'log.logger', operator: 'exists' as const },
+    result: 'ctx.log?.logger !== null',
+  },
+  {
+    condition: { field: 'log.logger', operator: 'notExists' as const },
+    result: 'ctx.log?.logger == null',
+  },
 ];
 
 describe('conditionToPainless', () => {

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

@@ -8,11 +8,13 @@
 import { isBoolean, isString } from 'lodash';
 import {
   AndCondition,
+  BinaryFilterCondition,
   Condition,
   conditionSchema,
   FilterCondition,
   filterConditionSchema,
   RerouteOrCondition,
+  UnaryFilterCondition,
 } from '../../../../common/types';
 
 function isFilterCondition(subject: any): subject is FilterCondition {
@@ -44,7 +46,7 @@ function encodeValue(value: string | number | boolean) {
   return value;
 }
 
-function toPainless(condition: FilterCondition) {
+function binaryToPainless(condition: BinaryFilterCondition) {
   switch (condition.operator) {
     case 'neq':
       return `${safePainlessField(condition)} != ${encodeValue(condition.value)}`;
@@ -67,9 +69,25 @@ function toPainless(condition: FilterCondition) {
   }
 }
 
+function unaryToPainless(condition: UnaryFilterCondition) {
+  switch (condition.operator) {
+    case 'notExists':
+      return `${safePainlessField(condition)} == null`;
+    default:
+      return `${safePainlessField(condition)} !== null`;
+  }
+}
+
+function isUnaryFilterCondition(subject: FilterCondition): subject is UnaryFilterCondition {
+  return !('value' in subject);
+}
+
 export function conditionToPainless(condition: Condition, nested = false): string {
   if (isFilterCondition(condition)) {
-    return `(${safePainlessField(condition)} !== null && ${toPainless(condition)})`;
+    if (isUnaryFilterCondition(condition)) {
+      return unaryToPainless(condition);
+    }
+    return `(${safePainlessField(condition)} !== null && ${binaryToPainless(condition)})`;
   }
   if (isAndCondition(condition)) {
     const and = condition.and.map((filter) => conditionToPainless(filter, true)).join(' && ');

x-pack/plugins/streams/server/lib/streams/index_templates/generate_index_template.ts

@@ -27,6 +27,7 @@ export function generateIndexTemplate(id: string) {
     },
     data_stream: {
       hidden: false,
+      failure_store: true,
     },
     template: {
       settings: {