[8.18] FTR - fix esArchiver duplicate doc ingestion (#229457)

src/platform/packages/shared/kbn-es-archiver/src/actions/load.ts

@@ -79,6 +79,7 @@ export async function loadAction({
   const stats = createStats(name, log);
   const files = prioritizeMappings(await readDirectory(inputDir));
   const kibanaPluginIds = await kbnClient.plugins.getEnabledIds();
+  const targetsWithoutIdGeneration: string[] = [];
 
   // a single stream that emits records from all archive files, in
   // order, so that createIndexStream can track the state of indexes
@@ -107,8 +108,16 @@ export async function loadAction({
       docsOnly,
       isArchiveInExceptionList,
       log,
+      targetsWithoutIdGeneration,
     }),
-    createIndexDocRecordsStream(client, stats, progress, useCreate, performance),
+    createIndexDocRecordsStream(
+      client,
+      stats,
+      progress,
+      useCreate,
+      performance,
+      targetsWithoutIdGeneration
+    ),
   ]);
 
   progress.deactivate();

src/platform/packages/shared/kbn-es-archiver/src/lib/docs/index_doc_records_stream.ts

@@ -10,6 +10,7 @@
 import type { Client } from '@elastic/elasticsearch';
 import AggregateError from 'aggregate-error';
 import { Writable } from 'stream';
+import { v4 as uuidv4 } from 'uuid';
 import { Stats } from '../stats';
 import { Progress } from '../progress';
 import { ES_CLIENT_HEADERS } from '../../client_headers';
@@ -24,7 +25,8 @@ export function createIndexDocRecordsStream(
   stats: Stats,
   progress: Progress,
   useCreate: boolean = false,
-  performance?: LoadActionPerfOptions
+  performance?: LoadActionPerfOptions,
+  targetsWithoutIdGeneration: string[] = []
 ) {
   async function indexDocs(docs: any[]) {
     const operation = useCreate === true ? BulkOperation.Create : BulkOperation.Index;
@@ -39,10 +41,12 @@ export function createIndexDocRecordsStream(
           const body = doc.source;
           const op = doc.data_stream ? BulkOperation.Create : operation;
           const index = doc.data_stream || doc.index;
+          // generate id for valid targets if it doesn't exist yet
+          const id = targetsWithoutIdGeneration.includes(index) ? doc.id : doc.id ?? uuidv4();
           ops.set(body, {
             [op]: {
               _index: index,
-              _id: doc.id,
+              _id: id,
             },
           });
           return body;

src/platform/packages/shared/kbn-es-archiver/src/lib/indices/create_index_stream.ts

@@ -44,13 +44,15 @@ export function createCreateIndexStream({
   docsOnly = false,
   isArchiveInExceptionList = false,
   log,
+  targetsWithoutIdGeneration = [],
 }: {
   client: Client;
   stats: Stats;
   skipExisting?: boolean;
   docsOnly?: boolean;
   isArchiveInExceptionList?: boolean;
   log: ToolingLog;
+  targetsWithoutIdGeneration?: string[];
 }) {
   const skipDocsFromIndices = new Set();
 
@@ -110,6 +112,7 @@ export function createCreateIndexStream({
         }
       );
       stats.createdDataStream(dataStream, template.name, { template });
+      targetsWithoutIdGeneration.push(dataStream);
     } catch (err) {
       if (err?.meta?.body?.error?.type !== 'resource_already_exists_exception' || attempts >= 3) {
         throw err;
@@ -188,6 +191,9 @@ export function createCreateIndexStream({
         }
 
         stats.createdIndex(index, { settings });
+        if (settings?.index?.mode === 'time_series') {
+          targetsWithoutIdGeneration.push(index);
+        }
       } catch (err) {
         if (
           err?.body?.error?.reason?.includes('index exists with the same name as the alias') &&