Migrationsv2: limit batch sizes to migrations.batchSizeBytes (= 100mb by default) (#109540)

* Fix logging for existing integration test

* First stab at limiting batches to batchSizeBytes

* Fix tests

* Fix batch size calculation, NDJSON needs to be terminated by an empty line

* Integration tests

* Fix type failures

* rename migration integration tests and log files to be consistent & more descriptive

* Review feedback

* Remove duplication of fatal error reasons

* migrations.maxBatchSizeBytes to docker environment vars

* docs for migrations.maxBatchSizeBytes

Author: rudolf

docs/setup/settings.asciidoc

@@ -406,7 +406,10 @@ override this parameter to use their own Tile Map Service. For example:
 `"https://tiles.elastic.co/v2/default/{z}/{x}/{y}.png?elastic_tile_service_tos=agree&my_app_name=kibana"`
 
 | `migrations.batchSize:`
- | Defines the number of documents migrated at a time. The higher the value, the faster the Saved Objects migration process performs at the cost of higher memory consumption. If the migration fails due to a `circuit_breaking_exception`, set a smaller `batchSize` value. *Default: `1000`*
+ | Defines the number of documents migrated at a time. The higher the value, the faster the Saved Objects migration process performs at the cost of higher memory consumption. If upgrade migrations results in {kib} crashing with an out of memory exception or fails due to an Elasticsearch `circuit_breaking_exception`, use a smaller `batchSize` value to reduce the memory pressure. *Default: `1000`*
+
+ | `migrations.maxBatchSizeBytes:`
+ | Defines the maximum payload size for indexing batches of upgraded saved objects to avoid migrations failing due to a 413 Request Entity Too Large response from Elasticsearch. This value should be lower than or equal to your Elasticsearch cluster's `http.max_content_length` configuration option. *Default: `100mb`*
 
 | `migrations.enableV2:`
  | experimental[]. Enables the new Saved Objects migration algorithm. For information about the migration algorithm, refer to <<upgrade-migrations>>. When `migrations v2` is stable, the setting will be removed in an upcoming release without any further notice. Setting the value to `false` causes {kib} to use the legacy migration algorithm, which shipped in 7.11 and earlier versions. *Default: `true`*

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

@@ -11,6 +11,7 @@ import { buildActiveMappings } from '../core';
 const { mergeTypes } = jest.requireActual('./kibana_migrator');
 import { SavedObjectsType } from '../../types';
 import { BehaviorSubject } from 'rxjs';
+import { ByteSizeValue } from '@kbn/config-schema';
 
 const defaultSavedObjectTypes: SavedObjectsType[] = [
   {
@@ -37,6 +38,7 @@ const createMigrator = (
     kibanaVersion: '8.0.0-testing',
     soMigrationsConfig: {
       batchSize: 100,
+      maxBatchSizeBytes: ByteSizeValue.parse('30kb'),
       scrollDuration: '15m',
       pollInterval: 1500,
       skip: false,

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

@@ -15,6 +15,7 @@ import { loggingSystemMock } from '../../../logging/logging_system.mock';
 import { SavedObjectTypeRegistry } from '../../saved_objects_type_registry';
 import { SavedObjectsType } from '../../types';
 import { DocumentMigrator } from '../core/document_migrator';
+import { ByteSizeValue } from '@kbn/config-schema';
 jest.mock('../core/document_migrator', () => {
   return {
     // Create a mock for spying on the constructor
@@ -396,6 +397,7 @@ const mockOptions = ({ enableV2 }: { enableV2: boolean } = { enableV2: false })
     } as KibanaMigratorOptions['kibanaConfig'],
     soMigrationsConfig: {
       batchSize: 20,
+      maxBatchSizeBytes: ByteSizeValue.parse('20mb'),
       pollInterval: 20000,
       scrollDuration: '10m',
       skip: false,

src/core/server/saved_objects/migrationsv2/README.md

@@ -316,7 +316,10 @@ completed this step:
  - temp index has a write block
  - temp index is not found
 ### New control state
+1. If `currentBatch` is the last batch in `transformedDocBatches`
   → `REINDEX_SOURCE_TO_TEMP_READ`
+2. If there are more batches left in `transformedDocBatches`
+  → `REINDEX_SOURCE_TO_TEMP_INDEX_BULK`
 
 ## REINDEX_SOURCE_TO_TEMP_CLOSE_PIT
 ### Next action

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

@@ -23,6 +23,27 @@ import type {
   IndexNotFound,
 } from './index';
 
+/**
+ * Given a document and index, creates a valid body for the Bulk API.
+ */
+export const createBulkOperationBody = (doc: SavedObjectsRawDoc, index: string) => {
+  return [
+    {
+      index: {
+        _index: index,
+        _id: doc._id,
+        // overwrite existing documents
+        op_type: 'index',
+        // use optimistic concurrency control to ensure that outdated
+        // documents are only overwritten once with the latest version
+        if_seq_no: doc._seq_no,
+        if_primary_term: doc._primary_term,
+      },
+    },
+    doc._source,
+  ];
+};
+
 /** @internal */
 export interface BulkOverwriteTransformedDocumentsParams {
   client: ElasticsearchClient;
@@ -47,6 +68,10 @@ export const bulkOverwriteTransformedDocuments = ({
   | RequestEntityTooLargeException,
   'bulk_index_succeeded'
 > => () => {
+  const body = transformedDocs.flatMap((doc) => {
+    return createBulkOperationBody(doc, index);
+  });
+
   return client
     .bulk({
       // Because we only add aliases in the MARK_VERSION_INDEX_READY step we
@@ -60,23 +85,7 @@ export const bulkOverwriteTransformedDocuments = ({
       wait_for_active_shards: WAIT_FOR_ALL_SHARDS_TO_BE_ACTIVE,
       refresh,
       filter_path: ['items.*.error'],
-      body: transformedDocs.flatMap((doc) => {
-        return [
-          {
-            index: {
-              _index: index,
-              _id: doc._id,
-              // overwrite existing documents
-              op_type: 'index',
-              // use optimistic concurrency control to ensure that outdated
-              // documents are only overwritten once with the latest version
-              if_seq_no: doc._seq_no,
-              if_primary_term: doc._primary_term,
-            },
-          },
-          doc._source,
-        ];
-      }),
+      body,
     })
     .then((res) => {
       // Filter out version_conflict_engine_exception since these just mean

src/core/server/saved_objects/migrationsv2/initial_state.test.ts

@@ -6,6 +6,7 @@
  * Side Public License, v 1.
  */
 
+import { ByteSizeValue } from '@kbn/config-schema';
 import * as Option from 'fp-ts/Option';
 import { SavedObjectsMigrationConfigType } from '../saved_objects_config';
 import { SavedObjectTypeRegistry } from '../saved_objects_type_registry';
@@ -21,6 +22,7 @@ describe('createInitialState', () => {
   const migrationsConfig = ({
     retryAttempts: 15,
     batchSize: 1000,
+    maxBatchSizeBytes: ByteSizeValue.parse('100mb'),
   } as unknown) as SavedObjectsMigrationConfigType;
   it('creates the initial state for the model based on the passed in parameters', () => {
     expect(
@@ -37,6 +39,7 @@ describe('createInitialState', () => {
       })
     ).toEqual({
       batchSize: 1000,
+      maxBatchSizeBytes: ByteSizeValue.parse('100mb').getValueInBytes(),
       controlState: 'INIT',
       currentAlias: '.kibana_task_manager',
       excludeFromUpgradeFilterHooks: {},

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

@@ -82,6 +82,7 @@ export const createInitialState = ({
     retryDelay: 0,
     retryAttempts: migrationsConfig.retryAttempts,
     batchSize: migrationsConfig.batchSize,
+    maxBatchSizeBytes: migrationsConfig.maxBatchSizeBytes.getValueInBytes(),
     logs: [],
     unusedTypesQuery: excludeUnusedTypesQuery,
     knownTypes,

src/core/server/saved_objects/migrationsv2/integration_tests/migration_7.7.2_xpack_100k.test.ts renamed to src/core/server/saved_objects/migrationsv2/integration_tests/7.7.2_xpack_100k.test.ts

@@ -17,7 +17,7 @@ import { InternalCoreStart } from '../../../internal_types';
 import { Root } from '../../../root';
 
 const kibanaVersion = Env.createDefault(REPO_ROOT, getEnvOptions()).packageInfo.version;
-const logFilePath = path.join(__dirname, 'migration_test_kibana.log');
+const logFilePath = path.join(__dirname, '7.7.2_xpack_100k.log');
 
 async function removeLogFile() {
   // ignore errors if it doesn't exist
@@ -61,9 +61,12 @@ describe('migration from 7.7.2-xpack with 100k objects', () => {
               },
             },
           },
-          root: {
-            appenders: ['default', 'file'],
-          },
+          loggers: [
+            {
+              name: 'root',
+              appenders: ['file'],
+            },
+          ],
         },
       },
       {

src/core/server/saved_objects/migrationsv2/integration_tests/migration_7_13_0_failed_action_tasks.test.ts renamed to src/core/server/saved_objects/migrationsv2/integration_tests/7_13_0_failed_action_tasks.test.ts

@@ -12,7 +12,7 @@ import * as kbnTestServer from '../../../../test_helpers/kbn_server';
 import { Root } from '../../../root';
 import { ElasticsearchClient } from '../../../elasticsearch';
 
-const logFilePath = Path.join(__dirname, '7_13_failed_action_tasks_test.log');
+const logFilePath = Path.join(__dirname, '7_13_failed_action_tasks.log');
 
 async function removeLogFile() {
   // ignore errors if it doesn't exist

src/core/server/saved_objects/migrationsv2/integration_tests/migration_7_13_0_transform_failures.test.ts renamed to src/core/server/saved_objects/migrationsv2/integration_tests/7_13_0_transform_failures.test.ts

@@ -12,7 +12,7 @@ import Util from 'util';
 import * as kbnTestServer from '../../../../test_helpers/kbn_server';
 import { Root } from '../../../root';
 
-const logFilePath = Path.join(__dirname, '7_13_corrupt_transform_failures_test.log');
+const logFilePath = Path.join(__dirname, '7_13_corrupt_transform_failures.log');
 
 const asyncUnlink = Util.promisify(Fs.unlink);