🌊 [Streams] Add state based (snapshot) telemetry, report Stream State Errors (#236921)

## Summary

This PR implements the initial stats (daily snapshot) telemetry
collection for Streams. Also tracks the Streams State errors via EBT.

The PR also moves the StreamsStorageClient to a central location so that
consumers don't have to recreate their own versions of Storage
Adapter/Client.

### 📊 State-Based (Snapshot) Telemetry

#### Overview
Implements a basic telemetry collector that captures Streams usage
patterns ~and significant events metrics~.

#### Metrics Collected

<table border="1" cellpadding="5" cellspacing="0"
style="border-collapse: collapse; width: 100%;">
  <thead>
    <tr>
      <th width="200">Metric</th>
      <th width="100">Type</th>
      <th style="text-align: left; padding: 8px;">Description</th>
    </tr>
  </thead>
  <tbody>
    <tr>
<td colspan="3" style="padding: 8px;"><strong>Classic
Streams</strong></td>
    </tr>
    <tr>
      <td style="padding: 8px;"><code>changed_count</code></td>
<td style="padding: 8px; white-space: nowrap;"><code>long</code></td>
<td style="padding: 8px;">Number of classic streams modified from
default configuration. Calculated by counting streams in
<code>.kibana_streams</code> index (presence indicates
customization).</td>
    </tr>
    <tr>
      <td style="padding: 8px;"><code>with_processing_count</code></td>
<td style="padding: 8px; white-space: nowrap;"><code>long</code></td>
<td style="padding: 8px;">Number of classic streams with custom
processing steps. Counted by non-empty
<code>ingest.processing.steps</code> arrays.</td>
    </tr>
    <tr>
      <td style="padding: 8px;"><code>with_fields_count</code></td>
<td style="padding: 8px; white-space: nowrap;"><code>long</code></td>
<td style="padding: 8px;">Number of classic streams with custom field
overrides. Counted by non-empty
<code>ingest.classic.field_overrides</code> objects.</td>
    </tr>
    <tr>
<td style="padding: 8px;"><code>with_changed_retention_count</code></td>
<td style="padding: 8px; white-space: nowrap;"><code>long</code></td>
<td style="padding: 8px;">Number of classic streams with default
(inherited) retention changed.</td>
    </tr>
    <tr>
<td colspan="3" style="padding: 8px;"><strong>Wired
Streams</strong></td>
    </tr>
    <tr>
      <td style="padding: 8px;"><code>count</code></td>
<td style="padding: 8px; white-space: nowrap;"><code>long</code></td>
<td style="padding: 8px;">Total number of wired streams in the system.
Counted by streams matching WiredStream schema (excluding
GroupStreams).</td>
    </tr>
    <tr>
<td colspan="3" style="padding: 8px;"><strong> ~Significant Events~
</strong> removed</td>
    </tr>
  </tbody>
</table>

### 📊 Error-Based (EBT) Telemetry

The errors reported by Strreams state machine e.g. during
[`attemptChanges`](https://github.com/elastic/kibana/blob/5d4799fa165a1f85967c9b837b8c29bdd6f61611/x-pack/platform/plugins/shared/streams/server/lib/streams/state_management/state.ts#L69)
are now tracked via EBT.

### 📖 Notes

Only unit tests coverage is maintained currently. e2e tests for
telemetry could be implemented as a follow up or after the telemetry
matures.

### 🧪 How to test

<img width="3446" height="2080" alt="image"
src="https://github.com/user-attachments/assets/62f33aa7-497d-41c2-b993-12ba52b2ec6d"
/>

---------

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

Author: awahab07

x-pack/platform/plugins/private/telemetry_collection_xpack/schema/xpack_platform.json

@@ -6492,6 +6492,48 @@
         }
       }
     },
+    "streams": {
+      "properties": {
+        "classic_streams": {
+          "properties": {
+            "changed_count": {
+              "type": "long",
+              "_meta": {
+                "description": "Number of classic streams that have been modified from their default configuration. Calculated by presence in .kibana_streams (managed) index and type is \"classic\"."
+              }
+            },
+            "with_processing_count": {
+              "type": "long",
+              "_meta": {
+                "description": "Number of classic streams with custom processing steps configured. Calculated by counting streams with non-empty ingest.processing.steps arrays."
+              }
+            },
+            "with_fields_count": {
+              "type": "long",
+              "_meta": {
+                "description": "Number of classic streams with custom field overrides configured. Calculated by counting streams with non-empty ingest.classic.field_overrides objects."
+              }
+            },
+            "with_changed_retention_count": {
+              "type": "long",
+              "_meta": {
+                "description": "Number of classic streams with changed retention. Calculated by confirming stream lifecycle is not \"inherited\"."
+              }
+            }
+          }
+        },
+        "wired_streams": {
+          "properties": {
+            "count": {
+              "type": "long",
+              "_meta": {
+                "description": "Total number of wired streams in the system."
+              }
+            }
+          }
+        }
+      }
+    },
     "task_manager": {
       "properties": {
         "task_type_exclusion": {

x-pack/platform/plugins/shared/streams/server/lib/streams/client.ts

@@ -28,7 +28,7 @@ import {
 import { SecurityError } from './errors/security_error';
 import { StatusError } from './errors/status_error';
 import { LOGS_ROOT_STREAM_NAME, rootStreamDefinition } from './root_stream_definition';
-import type { StreamsStorageClient } from './service';
+import type { StreamsStorageClient } from './storage/streams_storage_client';
 import { State } from './state_management/state';
 import { checkAccess, checkAccessBulk } from './stream_crud';
 import { StreamsStatusConflictError } from './errors/streams_status_conflict_error';

x-pack/platform/plugins/shared/streams/server/lib/streams/create_streams_global_search_result_provider.ts

@@ -11,13 +11,13 @@ import type {
   GlobalSearchProviderResult,
   GlobalSearchResultProvider,
 } from '@kbn/global-search-plugin/server';
-import { StorageIndexAdapter } from '@kbn/storage-adapter';
 import { Streams } from '@kbn/streams-schema';
 import type { SearchHit } from '@kbn/es-types';
 import { OBSERVABILITY_STREAMS_ENABLE_GROUP_STREAMS } from '@kbn/management-settings-ids';
-import type { StreamsStorageClient, StreamsStorageSettings } from './service';
-import { streamsStorageSettings } from './service';
-import { migrateOnRead } from './helpers/migrate_on_read';
+import {
+  createStreamsStorageClient,
+  type StreamsStorageClient,
+} from './storage/streams_storage_client';
 import { checkAccessBulk } from './stream_crud';
 
 const streamTypes = ['classic stream', 'wired stream', 'group stream', 'stream'];
@@ -34,13 +34,7 @@ export function createStreamsGlobalSearchResultProvider(
         return from([]);
       }
 
-      const storageAdapter = new StorageIndexAdapter<
-        StreamsStorageSettings,
-        Streams.all.Definition
-      >(client.asInternalUser, logger, streamsStorageSettings, {
-        migrateSource: migrateOnRead,
-      });
-      const storageClient = storageAdapter.getClient();
+      const storageClient = createStreamsStorageClient(client.asInternalUser, logger);
 
       return from(findStreams({ term, types, maxResults, storageClient, client, core })).pipe(
         takeUntil(aborted$)

x-pack/platform/plugins/shared/streams/server/lib/streams/service.ts

@@ -6,32 +6,14 @@
  */
 
 import type { CoreSetup, KibanaRequest, Logger } from '@kbn/core/server';
-import type { IStorageClient, StorageSettings } from '@kbn/storage-adapter';
-import { StorageIndexAdapter, types } from '@kbn/storage-adapter';
-import type { Streams } from '@kbn/streams-schema';
 import { LockManagerService } from '@kbn/lock-manager';
 import type { StreamsPluginStartDependencies } from '../../types';
+import { createStreamsStorageClient } from './storage/streams_storage_client';
 import type { AssetClient } from './assets/asset_client';
 import type { QueryClient } from './assets/query/query_client';
 import { StreamsClient } from './client';
-import { migrateOnRead } from './helpers/migrate_on_read';
 import type { SystemClient } from './system/system_client';
 
-export const streamsStorageSettings = {
-  name: '.kibana_streams',
-  schema: {
-    properties: {
-      name: types.keyword(),
-      description: types.text(),
-      ingest: types.object({ enabled: false }),
-      group: types.object({ enabled: false }),
-    },
-  },
-} satisfies StorageSettings;
-
-export type StreamsStorageSettings = typeof streamsStorageSettings;
-export type StreamsStorageClient = IStorageClient<StreamsStorageSettings, Streams.all.Definition>;
-
 export class StreamsService {
   constructor(
     private readonly coreSetup: CoreSetup<StreamsPluginStartDependencies>,
@@ -57,23 +39,14 @@ export class StreamsService {
     const scopedClusterClient = coreStart.elasticsearch.client.asScoped(request);
     const isServerless = coreStart.elasticsearch.getCapabilities().serverless;
 
-    const storageAdapter = new StorageIndexAdapter<StreamsStorageSettings, Streams.all.Definition>(
-      scopedClusterClient.asInternalUser,
-      logger,
-      streamsStorageSettings,
-      {
-        migrateSource: migrateOnRead,
-      }
-    );
-
     return new StreamsClient({
       assetClient,
       queryClient,
       systemClient,
       logger,
       scopedClusterClient,
       lockManager: new LockManagerService(this.coreSetup, logger),
-      storageClient: storageAdapter.getClient(),
+      storageClient: createStreamsStorageClient(scopedClusterClient.asInternalUser, logger),
       request,
       isServerless,
       isDev: this.isDev,

x-pack/platform/plugins/shared/streams/server/lib/streams/state_management/types.ts

@@ -10,7 +10,7 @@ import type { Streams } from '@kbn/streams-schema';
 import type { LockManagerService } from '@kbn/lock-manager';
 import type { AssetClient } from '../assets/asset_client';
 import type { StreamsClient } from '../client';
-import type { StreamsStorageClient } from '../service';
+import type { StreamsStorageClient } from '../storage/streams_storage_client';
 import type { QueryClient } from '../assets/query/query_client';
 import type { SystemClient } from '../system/system_client';
 

x-pack/platform/plugins/shared/streams/server/lib/streams/helpers/migrate_on_read.test.ts renamed to x-pack/platform/plugins/shared/streams/server/lib/streams/storage/migrate_on_read.test.ts

@@ -0,0 +1,54 @@
+/*
+ * 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 {
+  StorageIndexAdapter,
+  type IStorageClient,
+  type StorageSettings,
+  types,
+} from '@kbn/storage-adapter';
+import type { Logger } from '@kbn/logging';
+import type { ElasticsearchClient } from '@kbn/core-elasticsearch-server';
+import type { Streams } from '@kbn/streams-schema';
+import { migrateOnRead } from './migrate_on_read';
+
+const streamsStorageSettings = {
+  name: '.kibana_streams',
+  schema: {
+    properties: {
+      name: types.keyword(),
+      description: types.text(),
+      ingest: types.object({ enabled: false }),
+      group: types.object({ enabled: false }),
+    },
+  },
+} satisfies StorageSettings;
+
+type StreamsStorageSettings = typeof streamsStorageSettings;
+export type StreamsStorageClient = IStorageClient<StreamsStorageSettings, Streams.all.Definition>;
+
+/**
+ * This ensures there's only one way to initialize a storage client for streams, with the proper
+ * settings and migration on read.
+ * @param esClient
+ * @param logger
+ */
+export function createStreamsStorageClient(
+  esClient: ElasticsearchClient,
+  logger: Logger
+): StreamsStorageClient {
+  const adapter = new StorageIndexAdapter<StreamsStorageSettings, Streams.all.Definition>(
+    esClient,
+    logger,
+    streamsStorageSettings,
+    {
+      migrateSource: migrateOnRead,
+    }
+  );
+
+  return adapter.getClient();
+}

x-pack/platform/plugins/shared/streams/server/lib/streams/helpers/migrate_on_read.ts renamed to x-pack/platform/plugins/shared/streams/server/lib/streams/storage/migrate_on_read.ts

@@ -6,8 +6,8 @@
  */
 
 import type { AnalyticsServiceSetup } from '@kbn/core-analytics-server';
-import type { StreamEndpointLatencyProps } from './types';
-import { STREAMS_ENDPOINT_LATENCY_EVENT } from './constants';
+import type { StreamEndpointLatencyProps, StreamsStateErrorProps } from './types';
+import { STREAMS_ENDPOINT_LATENCY_EVENT, STREAMS_STATE_ERROR_EVENT } from './constants';
 
 const LATENCY_TRACKING_ENDPOINT_ALLOW_LIST = [
   'POST /api/streams/{name}/processing/_simulate 2023-10-31',
@@ -23,7 +23,7 @@ const LATENCY_TRACKING_ENDPOINT_ALLOW_LIST = [
   'POST /api/streams/_resync 2023-10-31',
 ];
 
-export class StreamsTelemetryClient {
+export class EbtTelemetryClient {
   constructor(private readonly analytics: AnalyticsServiceSetup) {}
 
   public startTrackingEndpointLatency(
@@ -40,4 +40,17 @@ export class StreamsTelemetryClient {
       });
     };
   }
+
+  public reportStreamsStateError(error: Error & { statusCode: number }) {
+    const errorData: StreamsStateErrorProps = {
+      error: {
+        name: error.name,
+        message: error.message,
+        stack_trace: error.stack,
+      },
+      status_code: error.statusCode,
+    };
+
+    this.analytics.reportEvent(STREAMS_STATE_ERROR_EVENT, errorData);
+  }
 }