🌊 Streams: Fix migration on read for stream definitions (elastic#220878)

Fixes elastic#220395

This is mostly meant to address the existing issue - there is a separate
issue to track a more long-term solution to the issue.

Here, we make sure that an empty description is added when a stream
definition is loaded from source - it might not be available because the
description property was added after streams was initially released.

It also adds an API integration to test for this - this should notify us
if the problem ever occurs again.

---------

Co-authored-by: kibanamachine <42973632+kibanamachine@users.noreply.github.com>

Author: flash1293

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

@@ -26,6 +26,7 @@ import { SecurityError } from './errors/security_error';
 import { State } from './state_management/state';
 import { StatusError } from './errors/status_error';
 import { ASSET_ID, ASSET_TYPE } from './assets/fields';
+import { migrateOnRead } from './helpers/migrate_on_read';
 
 interface AcknowledgeResponse<TResult extends Result> {
   acknowledged: true;
@@ -339,7 +340,7 @@ export class StreamsClient {
     try {
       const response = await this.dependencies.storageClient.get({ id: name });
 
-      const streamDefinition = response._source!;
+      const streamDefinition = migrateOnRead(response._source!);
 
       Streams.all.Definition.asserts(streamDefinition);
 

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

@@ -0,0 +1,18 @@
+/*
+ * 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 { Streams } from '@kbn/streams-schema';
+
+export function migrateOnRead(definition: Streams.all.Definition): Streams.all.Definition {
+  if (typeof definition.description !== 'string') {
+    return {
+      ...definition,
+      description: '',
+    };
+  }
+  return definition;
+}

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

@@ -11,12 +11,14 @@ import type { StreamActiveRecord } from './stream_active_record';
 import { UnwiredStream } from '../streams/unwired_stream';
 import { WiredStream } from '../streams/wired_stream';
 import { GroupStream } from '../streams/group_stream';
+import { migrateOnRead } from '../../helpers/migrate_on_read';
 
 // This should be the only thing that knows about the various stream types
 export function streamFromDefinition(
   definition: Streams.all.Definition,
   dependencies: StateDependencies
 ): StreamActiveRecord {
+  definition = migrateOnRead(definition);
   if (Streams.WiredStream.Definition.is(definition)) {
     return new WiredStream(definition, dependencies);
   } else if (Streams.UnwiredStream.Definition.is(definition)) {

x-pack/test/api_integration/deployment_agnostic/apis/observability/streams/index.ts

@@ -23,5 +23,6 @@ export default function ({ loadTestFile }: DeploymentAgnosticFtrProviderContext)
     loadTestFile(require.resolve('./queries'));
     loadTestFile(require.resolve('./discover'));
     loadTestFile(require.resolve('./content'));
+    loadTestFile(require.resolve('./migration_on_read'));
   });
 }

x-pack/test/api_integration/deployment_agnostic/apis/observability/streams/migration_on_read.ts

@@ -0,0 +1,118 @@
+/*
+ * 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 expect from '@kbn/expect';
+import { Streams } from '@kbn/streams-schema';
+import { DeploymentAgnosticFtrProviderContext } from '../../../ftr_provider_context';
+import { disableStreams, enableStreams, indexDocument } from './helpers/requests';
+import {
+  StreamsSupertestRepositoryClient,
+  createStreamsRepositoryAdminClient,
+} from './helpers/repository_client';
+
+const TEST_STREAM_NAME = 'logs-test-default';
+
+const streamDefinition = {
+  name: TEST_STREAM_NAME,
+  ingest: {
+    lifecycle: {
+      ilm: {
+        policy: 'logs-default',
+      },
+    },
+    processing: [
+      {
+        grok: {
+          field: 'message',
+          patterns: [
+            '%{TIMESTAMP_ISO8601:inner_timestamp} %{LOGLEVEL:log.level} %{GREEDYDATA:message2}',
+          ],
+          if: { always: {} },
+        },
+      },
+    ],
+    unwired: {},
+  },
+};
+
+const expectedStreamsResponse: Streams.UnwiredStream.Definition = {
+  name: TEST_STREAM_NAME,
+  description: '',
+  ingest: {
+    lifecycle: {
+      ilm: {
+        policy: 'logs-default',
+      },
+    },
+    processing: [
+      {
+        grok: {
+          field: 'message',
+          patterns: [
+            '%{TIMESTAMP_ISO8601:inner_timestamp} %{LOGLEVEL:log.level} %{GREEDYDATA:message2}',
+          ],
+          if: { always: {} },
+        },
+      },
+    ],
+    unwired: {},
+  },
+};
+
+export default function ({ getService }: DeploymentAgnosticFtrProviderContext) {
+  const roleScopedSupertest = getService('roleScopedSupertest');
+  let apiClient: StreamsSupertestRepositoryClient;
+  const esClient = getService('es');
+
+  // This test verifies that it's still possible to read an existing stream definition without
+  // error. If it fails, it indicates that the migration logic is not working as expected.
+  describe('read existing stream definition format', () => {
+    before(async () => {
+      apiClient = await createStreamsRepositoryAdminClient(roleScopedSupertest);
+      await enableStreams(apiClient);
+    });
+
+    after(async () => {
+      await disableStreams(apiClient);
+    });
+
+    it('should read and return existing orphaned classic stream', async () => {
+      await esClient.index({
+        index: '.kibana_streams-000001',
+        id: TEST_STREAM_NAME,
+        document: streamDefinition,
+      });
+
+      // Refresh the index to make the document searchable
+      await esClient.indices.refresh({ index: '.kibana_streams-000001' });
+      const getResponse = await apiClient.fetch('GET /api/streams/{name} 2023-10-31', {
+        params: {
+          path: { name: TEST_STREAM_NAME },
+        },
+      });
+
+      expect(getResponse.status).to.eql(200);
+      expect(getResponse.body.stream).to.eql(expectedStreamsResponse);
+    });
+
+    it('should read and return existing regular classic stream', async () => {
+      const doc = {
+        message: '2023-01-01T00:00:10.000Z error test',
+      };
+      const response = await indexDocument(esClient, TEST_STREAM_NAME, doc);
+      expect(response.result).to.eql('created');
+      const getResponse = await apiClient.fetch('GET /api/streams/{name} 2023-10-31', {
+        params: {
+          path: { name: TEST_STREAM_NAME },
+        },
+      });
+
+      expect(getResponse.status).to.eql(200);
+      expect(getResponse.body.stream).to.eql(expectedStreamsResponse);
+    });
+  });
+}