🌊 Streams: Enable failure store for wired streams (#234066)

Closes elastic/streams-program#304 by enabling
failure store for all wired streams.

This also means that documents that violate the rules of wired streams
will go to the failure store (e.g. docs going directly to a child
stream)

Author: flash1293

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

@@ -35,14 +35,18 @@ export function generateIndexTemplate(name: string) {
     },
     data_stream: {
       hidden: false,
-      failure_store: false,
     },
     template: {
       settings: {
         index: {
           default_pipeline: getProcessingPipelineName(name),
         },
       },
+      data_stream_options: {
+        failure_store: {
+          enabled: true,
+        },
+      },
       mappings: {
         properties: {
           'stream.name': {

x-pack/platform/test/api_integration_deployment_agnostic/apis/streams/helpers/requests.ts

@@ -9,7 +9,7 @@ import type { Readable } from 'stream';
 import type { Client } from '@elastic/elasticsearch';
 import type { JsonObject } from '@kbn/utility-types';
 import expect from '@kbn/expect';
-import type { SearchTotalHits } from '@elastic/elasticsearch/lib/api/types';
+import type { SearchTotalHits, Refresh } from '@elastic/elasticsearch/lib/api/types';
 import type { Streams } from '@kbn/streams-schema';
 import type { ClientRequestParamsOf } from '@kbn/server-route-repository-utils';
 import type { StreamsRouteRepository } from '@kbn/streams-plugin/server';
@@ -24,8 +24,13 @@ export async function disableStreams(client: StreamsSupertestRepositoryClient) {
   await client.fetch('POST /api/streams/_disable 2023-10-31').expect(200);
 }
 
-export async function indexDocument(esClient: Client, index: string, document: JsonObject) {
-  const response = await esClient.index({ index, document, refresh: 'wait_for' });
+export async function indexDocument(
+  esClient: Client,
+  index: string,
+  document: JsonObject,
+  refresh: Refresh = 'wait_for'
+) {
+  const response = await esClient.index({ index, document, refresh });
   return response;
 }
 

x-pack/platform/test/api_integration_deployment_agnostic/apis/streams/root_stream.ts

@@ -83,8 +83,6 @@ export default function ({ getService }: DeploymentAgnosticFtrProviderContext) {
   const roleScopedSupertest = getService('roleScopedSupertest');
   let apiClient: StreamsSupertestRepositoryClient;
   const esClient = getService('es');
-  const config = getService('config');
-  const isServerless = !!config.get('serverless');
 
   describe('Root stream', () => {
     before(async () => {
@@ -190,18 +188,9 @@ export default function ({ getService }: DeploymentAgnosticFtrProviderContext) {
         '@timestamp': '2024-01-01T00:00:20.000Z',
         message: 'test',
       };
-      let threw = false;
-      try {
-        await indexDocument(esClient, 'logs.gcpcloud', doc);
-      } catch (e) {
-        threw = true;
-        if (isServerless) {
-          expect(e.message).to.contain('stream.name is not set properly');
-        } else {
-          expect(e.message).to.contain('Direct writes to child streams are prohibited');
-        }
-      }
-      expect(threw).to.be(true);
+      const response = await indexDocument(esClient, 'logs.gcpcloud', doc, false);
+      // @ts-expect-error failure_store is not in the types, but in the actual response
+      expect(response.failure_store).to.be('used');
     });
   });
 }