[SoMigV2] Fail fast if unknown document types are present in the source index

src/core/server/saved_objects/migrations/kibana/kibana_migrator.ts

@@ -192,6 +192,7 @@ export class KibanaMigrator {
               migrationVersionPerType: this.documentMigrator.migrationVersion,
               indexPrefix: index,
               migrationsConfig: this.soMigrationsConfig,
+              typeRegistry: this.typeRegistry,
             });
           },
         };

src/core/server/saved_objects/migrationsv2/actions/check_for_unknown_docs.ts

@@ -0,0 +1,79 @@
+/*
+ * 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 and the Server Side Public License, v 1; you may not use this file except
+ * in compliance with, at your election, the Elastic License 2.0 or the Server
+ * Side Public License, v 1.
+ */
+
+import * as Either from 'fp-ts/lib/Either';
+import * as TaskEither from 'fp-ts/lib/TaskEither';
+import { estypes } from '@elastic/elasticsearch';
+import type { SavedObjectsRawDocSource } from '../../serialization';
+import { ElasticsearchClient } from '../../../elasticsearch';
+import {
+  catchRetryableEsClientErrors,
+  RetryableEsClientError,
+} from './catch_retryable_es_client_errors';
+
+/** @internal */
+export interface CheckForUnknownDocsParams {
+  client: ElasticsearchClient;
+  indexName: string;
+  unusedTypesQuery: estypes.QueryDslQueryContainer;
+  knownTypes: string[];
+}
+
+/** @internal */
+export interface CheckForUnknownDocsResponseDoc {
+  id: string;
+  type: string;
+}
+
+/** @internal */
+export interface CheckForUnknownDocsResponse {
+  unknownDocs: CheckForUnknownDocsResponseDoc[];
+}
+
+export const checkForUnknownDocs = ({
+  client,
+  indexName,
+  unusedTypesQuery,
+  knownTypes,
+}: CheckForUnknownDocsParams): TaskEither.TaskEither<
+  RetryableEsClientError,
+  CheckForUnknownDocsResponse
+> => () => {
+  const query = createUnknownDocQuery(unusedTypesQuery, knownTypes);
+
+  return client
+    .search<SavedObjectsRawDocSource>({
+      index: indexName,
+      body: {
+        query,
+      },
+    })
+    .then((response) => {
+      const { hits } = response.body.hits;
+      return Either.right({
+        unknownDocs: hits.map((hit) => ({ id: hit._id, type: hit._source?.type ?? 'undefined' })),
+      });
+    })
+    .catch(catchRetryableEsClientErrors);
+};
+
+const createUnknownDocQuery = (
+  unusedTypesQuery: estypes.QueryDslQueryContainer,
+  knownTypes: string[]
+): estypes.QueryDslQueryContainer => {
+  return {
+    bool: {
+      must: unusedTypesQuery,
+      must_not: knownTypes.map((type) => ({
+        term: {
+          type,
+        },
+      })),
+    },
+  };
+};

src/core/server/saved_objects/migrationsv2/actions/index.ts

@@ -60,12 +60,14 @@ export type { ReindexResponse, ReindexParams } from './reindex';
 export { reindex } from './reindex';
 
 import type { IncompatibleMappingException } from './wait_for_reindex_task';
+
 export { waitForReindexTask } from './wait_for_reindex_task';
 
 export type { VerifyReindexParams } from './verify_reindex';
 export { verifyReindex } from './verify_reindex';
 
 import type { AliasNotFound, RemoveIndexNotAConcreteIndex } from './update_aliases';
+
 export type { AliasAction, UpdateAliasesParams } from './update_aliases';
 export { updateAliases } from './update_aliases';
 
@@ -78,6 +80,13 @@ export type {
 } from './update_and_pickup_mappings';
 export { updateAndPickupMappings } from './update_and_pickup_mappings';
 
+export type {
+  CheckForUnknownDocsParams,
+  CheckForUnknownDocsResponse,
+  CheckForUnknownDocsResponseDoc,
+} from './check_for_unknown_docs';
+export { checkForUnknownDocs } from './check_for_unknown_docs';
+
 export { waitForPickupUpdatedMappingsTask } from './wait_for_pickup_updated_mappings_task';
 
 export type {
@@ -96,9 +105,11 @@ export interface IndexNotFound {
   type: 'index_not_found_exception';
   index: string;
 }
+
 export interface WaitForReindexTaskFailure {
   readonly cause: { type: string; reason: string };
 }
+
 export interface TargetIndexHadWriteBlock {
   type: 'target_index_had_write_block';
 }
@@ -108,6 +119,7 @@ export interface AcknowledgeResponse {
   acknowledged: boolean;
   shardsAcknowledged: boolean;
 }
+
 // Map of left response 'type' string -> response interface
 export interface ActionErrorTypeMap {
   wait_for_task_completion_timeout: WaitForTaskCompletionTimeout;

src/core/server/saved_objects/migrationsv2/actions/wait_for_index_status_yellow.ts

@@ -5,6 +5,7 @@
  * in compliance with, at your election, the Elastic License 2.0 or the Server
  * Side Public License, v 1.
  */
+
 import * as Either from 'fp-ts/lib/Either';
 import * as TaskEither from 'fp-ts/lib/TaskEither';
 import { ElasticsearchClient } from '../../../elasticsearch';

src/core/server/saved_objects/migrationsv2/index.ts

@@ -13,9 +13,11 @@ import type { SavedObjectsMigrationVersion } from '../types';
 import type { TransformRawDocs } from './types';
 import { MigrationResult } from '../migrations/core';
 import { next } from './next';
-import { createInitialState, model } from './model';
+import { model } from './model';
+import { createInitialState } from './initial_state';
 import { migrationStateActionMachine } from './migrations_state_action_machine';
 import { SavedObjectsMigrationConfigType } from '../saved_objects_config';
+import type { ISavedObjectTypeRegistry } from '../saved_objects_type_registry';
 
 /**
  * Migrates the provided indexPrefix index using a resilient algorithm that is
@@ -32,6 +34,7 @@ export async function runResilientMigrator({
   migrationVersionPerType,
   indexPrefix,
   migrationsConfig,
+  typeRegistry,
 }: {
   client: ElasticsearchClient;
   kibanaVersion: string;
@@ -42,6 +45,7 @@ export async function runResilientMigrator({
   migrationVersionPerType: SavedObjectsMigrationVersion;
   indexPrefix: string;
   migrationsConfig: SavedObjectsMigrationConfigType;
+  typeRegistry: ISavedObjectTypeRegistry;
 }): Promise<MigrationResult> {
   const initialState = createInitialState({
     kibanaVersion,
@@ -50,6 +54,7 @@ export async function runResilientMigrator({
     migrationVersionPerType,
     indexPrefix,
     migrationsConfig,
+    typeRegistry,
   });
   return migrationStateActionMachine({
     initialState,