[8.x] [Streams] Introducing the new Streams plugin (#198713) (#201723)

# Backport

This will backport the following commits from `main` to `8.x`:
- [[Streams] Introducing the new Streams plugin
(#198713)](#198713)

<!--- Backport version: 7.3.2 -->

### Questions ?
Please refer to the [Backport tool
documentation](https://github.com/sqren/backport)

<!--BACKPORT {commits} BACKPORT-->

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

docs/developer/plugin-list.asciidoc

@@ -909,6 +909,10 @@ routes, etc.
 |The stack_connectors plugin provides connector types shipped with Kibana, built on top of the framework provided in the actions plugin.
 
 
+|{kib-repo}blob/{branch}/x-pack/plugins/streams/README.md[streams]
+|This plugin provides an interface to manage streams
+
+
 |{kib-repo}blob/{branch}/x-pack/plugins/observability_solution/synthetics/README.md[synthetics]
 |The purpose of this plugin is to provide users of Heartbeat more visibility of what's happening
 in their infrastructure.

package.json

@@ -934,6 +934,7 @@
     "@kbn/status-plugin-a-plugin": "link:test/server_integration/plugins/status_plugin_a",
     "@kbn/status-plugin-b-plugin": "link:test/server_integration/plugins/status_plugin_b",
     "@kbn/std": "link:packages/kbn-std",
+    "@kbn/streams-plugin": "link:x-pack/plugins/streams",
     "@kbn/synthetics-plugin": "link:x-pack/plugins/observability_solution/synthetics",
     "@kbn/synthetics-private-location": "link:x-pack/packages/kbn-synthetics-private-location",
     "@kbn/task-manager-fixture-plugin": "link:x-pack/test/alerting_api_integration/common/plugins/task_manager_fixture",

packages/kbn-optimizer/limits.yml

@@ -161,6 +161,7 @@ pageLoadAssetSize:
   spaces: 57868
   stackAlerts: 58316
   stackConnectors: 67227
+  streams: 16742
   synthetics: 55971
   telemetry: 51957
   telemetryManagementSection: 38586

tsconfig.base.json

@@ -1844,6 +1844,8 @@
       "@kbn/stdio-dev-helpers/*": ["packages/kbn-stdio-dev-helpers/*"],
       "@kbn/storybook": ["packages/kbn-storybook"],
       "@kbn/storybook/*": ["packages/kbn-storybook/*"],
+      "@kbn/streams-plugin": ["x-pack/plugins/streams"],
+      "@kbn/streams-plugin/*": ["x-pack/plugins/streams/*"],
       "@kbn/synthetics-e2e": ["x-pack/plugins/observability_solution/synthetics/e2e"],
       "@kbn/synthetics-e2e/*": ["x-pack/plugins/observability_solution/synthetics/e2e/*"],
       "@kbn/synthetics-plugin": ["x-pack/plugins/observability_solution/synthetics"],

x-pack/plugins/streams/README.md

@@ -0,0 +1,3 @@
+# Streams Plugin
+
+This plugin provides an interface to manage streams

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

@@ -0,0 +1,30 @@
+/*
+ * 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 { schema, TypeOf } from '@kbn/config-schema';
+
+export const configSchema = schema.object({});
+
+export type StreamsConfig = TypeOf<typeof configSchema>;
+
+/**
+ * The following map is passed to the server plugin setup under the
+ * exposeToBrowser: option, and controls which of the above config
+ * keys are allow-listed to be available in the browser config.
+ *
+ * NOTE: anything exposed here will be visible in the UI dev tools,
+ * and therefore MUST NOT be anything that is sensitive information!
+ */
+export const exposeToBrowserConfig = {} as const;
+
+type ValidKeys = keyof {
+  [K in keyof typeof exposeToBrowserConfig as (typeof exposeToBrowserConfig)[K] extends true
+    ? K
+    : never]: true;
+};
+
+export type StreamsPublicConfig = Pick<StreamsConfig, ValidKeys>;

x-pack/plugins/streams/common/constants.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 const ASSET_VERSION = 1;
+export const STREAMS_INDEX = '.kibana_streams';

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

@@ -0,0 +1,91 @@
+/*
+ * 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';
+
+const stringOrNumberOrBoolean = z.union([z.string(), z.number(), z.boolean()]);
+
+export const filterConditionSchema = z.object({
+  field: z.string(),
+  operator: z.enum(['eq', 'neq', 'lt', 'lte', 'gt', 'gte', 'contains', 'startsWith', 'endsWith']),
+  value: stringOrNumberOrBoolean,
+});
+
+export type FilterCondition = z.infer<typeof filterConditionSchema>;
+
+export interface AndCondition {
+  and: Condition[];
+}
+
+export interface RerouteOrCondition {
+  or: Condition[];
+}
+
+export type Condition = FilterCondition | AndCondition | RerouteOrCondition | undefined;
+
+export const conditionSchema: z.ZodType<Condition> = z.lazy(() =>
+  z.union([
+    filterConditionSchema,
+    z.object({ and: z.array(conditionSchema) }),
+    z.object({ or: z.array(conditionSchema) }),
+  ])
+);
+
+export const grokProcessingDefinitionSchema = z.object({
+  type: z.literal('grok'),
+  field: z.string(),
+  patterns: z.array(z.string()),
+  pattern_definitions: z.optional(z.record(z.string())),
+});
+
+export const dissectProcessingDefinitionSchema = z.object({
+  type: z.literal('dissect'),
+  field: z.string(),
+  pattern: z.string(),
+});
+
+export const processingDefinitionSchema = z.object({
+  condition: z.optional(conditionSchema),
+  config: z.discriminatedUnion('type', [
+    grokProcessingDefinitionSchema,
+    dissectProcessingDefinitionSchema,
+  ]),
+});
+
+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']),
+});
+
+export type FieldDefinition = z.infer<typeof fieldDefinitionSchema>;
+
+export const streamWithoutIdDefinitonSchema = z.object({
+  processing: z.array(processingDefinitionSchema).default([]),
+  fields: z.array(fieldDefinitionSchema).default([]),
+  children: z
+    .array(
+      z.object({
+        id: z.string(),
+        condition: conditionSchema,
+      })
+    )
+    .default([]),
+});
+
+export type StreamWithoutIdDefinition = z.infer<typeof streamDefinitonSchema>;
+
+export const streamDefinitonSchema = streamWithoutIdDefinitonSchema.extend({
+  id: z.string(),
+});
+
+export type StreamDefinition = z.infer<typeof streamDefinitonSchema>;
+
+export const streamDefinitonWithoutChildrenSchema = streamDefinitonSchema.omit({ children: true });
+
+export type StreamWithoutChildrenDefinition = z.infer<typeof streamDefinitonWithoutChildrenSchema>;

x-pack/plugins/streams/jest.config.js

@@ -0,0 +1,15 @@
+/*
+ * 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.
+ */
+
+module.exports = {
+  preset: '@kbn/test',
+  rootDir: '../../..',
+  roots: ['<rootDir>/x-pack/plugins/streams'],
+  coverageDirectory: '<rootDir>/target/kibana-coverage/jest/x-pack/plugins/streams',
+  coverageReporters: ['text', 'html'],
+  collectCoverageFrom: ['<rootDir>/x-pack/plugins/streams/{common,public,server}/**/*.{js,ts,tsx}'],
+};

x-pack/plugins/streams/kibana.jsonc

@@ -0,0 +1,28 @@
+{
+  "type": "plugin",
+  "id": "@kbn/streams-plugin",
+  "owner": "@simianhacker @flash1293 @dgieselaar",
+  "description": "A manager for Streams",
+  "group": "observability",
+  "visibility": "private",
+  "plugin": {
+    "id": "streams",
+    "configPath": ["xpack", "streams"],
+    "browser": true,
+    "server": true,
+    "requiredPlugins": [
+      "data",
+      "security",
+      "encryptedSavedObjects",
+      "usageCollection",
+      "licensing",
+      "taskManager"
+    ],
+    "optionalPlugins": [
+      "cloud",
+      "serverless"
+    ],
+    "requiredBundles": [
+    ]
+  }
+}

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

@@ -0,0 +1,13 @@
+/*
+ * 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 { PluginInitializer, PluginInitializerContext } from '@kbn/core/public';
+import { Plugin } from './plugin';
+
+export const plugin: PluginInitializer<{}, {}> = (context: PluginInitializerContext) => {
+  return new Plugin(context);
+};

x-pack/plugins/streams/public/plugin.ts

@@ -0,0 +1,32 @@
+/*
+ * 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 { CoreSetup, CoreStart, PluginInitializerContext } from '@kbn/core/public';
+import { Logger } from '@kbn/logging';
+
+import type { StreamsPublicConfig } from '../common/config';
+import { StreamsPluginClass, StreamsPluginSetup, StreamsPluginStart } from './types';
+
+export class Plugin implements StreamsPluginClass {
+  public config: StreamsPublicConfig;
+  public logger: Logger;
+
+  constructor(context: PluginInitializerContext<{}>) {
+    this.config = context.config.get();
+    this.logger = context.logger.get();
+  }
+
+  setup(core: CoreSetup<StreamsPluginStart>, pluginSetup: StreamsPluginSetup) {
+    return {};
+  }
+
+  start(core: CoreStart) {
+    return {};
+  }
+
+  stop() {}
+}

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

@@ -0,0 +1,16 @@
+/*
+ * 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 type { Plugin as PluginClass } from '@kbn/core/public';
+
+// eslint-disable-next-line @typescript-eslint/no-empty-interface
+export interface StreamsPluginSetup {}
+
+// eslint-disable-next-line @typescript-eslint/no-empty-interface
+export interface StreamsPluginStart {}
+
+export type StreamsPluginClass = PluginClass<{}, {}, StreamsPluginSetup, StreamsPluginStart>;

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

@@ -0,0 +1,19 @@
+/*
+ * 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 { PluginInitializerContext } from '@kbn/core-plugins-server';
+import { StreamsConfig } from '../common/config';
+import { StreamsPluginSetup, StreamsPluginStart, config } from './plugin';
+import { StreamsRouteRepository } from './routes';
+
+export type { StreamsConfig, StreamsPluginSetup, StreamsPluginStart, StreamsRouteRepository };
+export { config };
+
+export const plugin = async (context: PluginInitializerContext<StreamsConfig>) => {
+  const { StreamsPlugin } = await import('./plugin');
+  return new StreamsPlugin(context);
+};

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

@@ -0,0 +1,43 @@
+/*
+ * 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 {
+  ClusterPutComponentTemplateRequest,
+  MappingProperty,
+} from '@elastic/elasticsearch/lib/api/types';
+import { StreamDefinition } from '../../../../common/types';
+import { ASSET_VERSION } from '../../../../common/constants';
+import { logsSettings } from './logs_layer';
+import { isRoot } from '../helpers/hierarchy';
+import { getComponentTemplateName } from './name';
+
+export function generateLayer(
+  id: string,
+  definition: StreamDefinition
+): ClusterPutComponentTemplateRequest {
+  const properties: Record<string, MappingProperty> = {};
+  definition.fields.forEach((field) => {
+    properties[field.name] = {
+      type: field.type,
+    };
+  });
+  return {
+    name: getComponentTemplateName(id),
+    template: {
+      settings: isRoot(definition.id) ? logsSettings : {},
+      mappings: {
+        subobjects: false,
+        properties,
+      },
+    },
+    version: ASSET_VERSION,
+    _meta: {
+      managed: true,
+      description: `Default settings for the ${id} stream`,
+    },
+  };
+}

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

@@ -0,0 +1,23 @@
+/*
+ * 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 { IndicesIndexSettings } from '@elastic/elasticsearch/lib/api/types';
+
+export const logsSettings: IndicesIndexSettings = {
+  index: {
+    lifecycle: {
+      name: 'logs',
+    },
+    codec: 'best_compression',
+    mapping: {
+      total_fields: {
+        ignore_dynamic_beyond_limit: true,
+      },
+      ignore_malformed: true,
+    },
+  },
+};