[7.x] Add ability for types to define an excludeOnUpgrade hook (#106534…

…) (#106930)

* Add ability for types to define an excludeOnUpgrade hook (#106534)

* Add ability for types to define an exclude from upgrade hook

* Update test to use 7.13 fixture

* Add 404 handling to oldest action task query

* Update api docs

* Update and add unit tests

* Rename deleteOnUpgrade to excludeOnUpgrade

* Disable reporting browser download for integration test

* Update exports and TSDocs

* Disable reporting plugin entirely

* Typo in config key name

* Update api docs

* Fix eslint

* Add timeouts for hooks

* Make adjustments to getOldestIdleActionTask

* Add type name to log messages for failed hooks

* Update api docs

* Update fixture for 7.x

Co-authored-by: Josh Dover <[email protected]>

docs/development/core/public/kibana-plugin-core-public.kibanaexecutioncontext.md

@@ -18,6 +18,6 @@ export interface KibanaExecutionContext
 |  [description](./kibana-plugin-core-public.kibanaexecutioncontext.description.md) | <code>string</code> | human readable description. For example, a vis title, action name |
 |  [id](./kibana-plugin-core-public.kibanaexecutioncontext.id.md) | <code>string</code> | unique value to identify the source |
 |  [name](./kibana-plugin-core-public.kibanaexecutioncontext.name.md) | <code>string</code> | public name of a user-facing feature |
-|  [type](./kibana-plugin-core-public.kibanaexecutioncontext.type.md) | <code>string</code> | Kibana application initated an operation. Can be narrowed to an enum later. |
+|  [type](./kibana-plugin-core-public.kibanaexecutioncontext.type.md) | <code>string</code> | Kibana application initated an operation. |
 |  [url](./kibana-plugin-core-public.kibanaexecutioncontext.url.md) | <code>string</code> | in browser - url to navigate to a current page, on server - endpoint path, for task: task SO url |
 

docs/development/core/public/kibana-plugin-core-public.kibanaexecutioncontext.type.md

@@ -4,7 +4,7 @@
 
 ## KibanaExecutionContext.type property
 
-Kibana application initated an operation. Can be narrowed to an enum later.
+Kibana application initated an operation.
 
 <b>Signature:</b>
 

docs/development/core/server/kibana-plugin-core-server.md

@@ -329,6 +329,7 @@ The plugin integrates with the core system via lifecycle events: `setup`<!-- -->
 |  [SavedObjectsImportHook](./kibana-plugin-core-server.savedobjectsimporthook.md) | A hook associated with a specific saved object type, that will be invoked during the import process. The hook will have access to the objects of the registered type.<!-- -->Currently, the only supported feature for import hooks is to return warnings to be displayed in the UI when the import succeeds. The only interactions the hook can have with the import process is via the hook's response. Mutating the objects inside the hook's code will have no effect. |
 |  [SavedObjectsImportWarning](./kibana-plugin-core-server.savedobjectsimportwarning.md) | Composite type of all the possible types of import warnings.<!-- -->See [SavedObjectsImportSimpleWarning](./kibana-plugin-core-server.savedobjectsimportsimplewarning.md) and [SavedObjectsImportActionRequiredWarning](./kibana-plugin-core-server.savedobjectsimportactionrequiredwarning.md) for more details. |
 |  [SavedObjectsNamespaceType](./kibana-plugin-core-server.savedobjectsnamespacetype.md) | The namespace type dictates how a saved object can be interacted in relation to namespaces. Each type is mutually exclusive: \* single (default): This type of saved object is namespace-isolated, e.g., it exists in only one namespace. \* multiple: This type of saved object is shareable, e.g., it can exist in one or more namespaces. \* multiple-isolated: This type of saved object is namespace-isolated, e.g., it exists in only one namespace, but object IDs must be unique across all namespaces. This is intended to be an intermediate step when objects with a "single" namespace type are being converted to a "multiple" namespace type. In other words, objects with a "multiple-isolated" namespace type will be \*share-capable\*, but will not actually be shareable until the namespace type is changed to "multiple". \* agnostic: This type of saved object is global. |
+|  [SavedObjectTypeExcludeFromUpgradeFilterHook](./kibana-plugin-core-server.savedobjecttypeexcludefromupgradefilterhook.md) | If defined, allows a type to run a search query and return a query filter that may match any documents which may be excluded from the next migration upgrade process. Useful for cleaning up large numbers of old documents which are no longer needed and may slow the migration process.<!-- -->If this hook fails, the migration will proceed without these documents having been filtered out, so this should not be used as a guarantee that these documents have been deleted.<!-- -->Experimental and subject to change |
 |  [SavedObjectUnsanitizedDoc](./kibana-plugin-core-server.savedobjectunsanitizeddoc.md) | Describes Saved Object documents from Kibana &lt; 7.0.0 which don't have a <code>references</code> root property defined. This type should only be used in migrations. |
 |  [ScopeableRequest](./kibana-plugin-core-server.scopeablerequest.md) | A user credentials container. It accommodates the necessary auth credentials to impersonate the current user.<!-- -->See [KibanaRequest](./kibana-plugin-core-server.kibanarequest.md)<!-- -->. |
 |  [ServiceStatusLevel](./kibana-plugin-core-server.servicestatuslevel.md) | A convenience type that represents the union of each value in [ServiceStatusLevels](./kibana-plugin-core-server.servicestatuslevels.md)<!-- -->. |

docs/development/core/server/kibana-plugin-core-server.savedobjectstype.excludeonupgrade.md

@@ -0,0 +1,13 @@
+<!-- Do not edit this file. It is automatically generated by API Documenter. -->
+
+[Home](./index.md) &gt; [kibana-plugin-core-server](./kibana-plugin-core-server.md) &gt; [SavedObjectsType](./kibana-plugin-core-server.savedobjectstype.md) &gt; [excludeOnUpgrade](./kibana-plugin-core-server.savedobjectstype.excludeonupgrade.md)
+
+## SavedObjectsType.excludeOnUpgrade property
+
+If defined, allows a type to exclude unneeded documents from the migration process and effectively be deleted. See [SavedObjectTypeExcludeFromUpgradeFilterHook](./kibana-plugin-core-server.savedobjecttypeexcludefromupgradefilterhook.md) for more details.
+
+<b>Signature:</b>
+
+```typescript
+excludeOnUpgrade?: SavedObjectTypeExcludeFromUpgradeFilterHook;
+```

docs/development/core/server/kibana-plugin-core-server.savedobjectstype.md

@@ -52,6 +52,7 @@ Example after converting to a multi-namespace (shareable) type in 8.1:
 
 ```
 Note: migration function(s) can be optionally specified for any of these versions and will not interfere with the conversion process. |
+|  [excludeOnUpgrade](./kibana-plugin-core-server.savedobjectstype.excludeonupgrade.md) | <code>SavedObjectTypeExcludeFromUpgradeFilterHook</code> | If defined, allows a type to exclude unneeded documents from the migration process and effectively be deleted. See [SavedObjectTypeExcludeFromUpgradeFilterHook](./kibana-plugin-core-server.savedobjecttypeexcludefromupgradefilterhook.md) for more details. |
 |  [hidden](./kibana-plugin-core-server.savedobjectstype.hidden.md) | <code>boolean</code> | Is the type hidden by default. If true, repositories will not have access to this type unless explicitly declared as an <code>extraType</code> when creating the repository.<!-- -->See [createInternalRepository](./kibana-plugin-core-server.savedobjectsservicestart.createinternalrepository.md)<!-- -->. |
 |  [indexPattern](./kibana-plugin-core-server.savedobjectstype.indexpattern.md) | <code>string</code> | If defined, the type instances will be stored in the given index instead of the default one. |
 |  [management](./kibana-plugin-core-server.savedobjectstype.management.md) | <code>SavedObjectsTypeManagementDefinition&lt;Attributes&gt;</code> | An optional [saved objects management section](./kibana-plugin-core-server.savedobjectstypemanagementdefinition.md) definition for the type. |

docs/development/core/server/kibana-plugin-core-server.savedobjecttypeexcludefromupgradefilterhook.md

@@ -0,0 +1,19 @@
+<!-- Do not edit this file. It is automatically generated by API Documenter. -->
+
+[Home](./index.md) &gt; [kibana-plugin-core-server](./kibana-plugin-core-server.md) &gt; [SavedObjectTypeExcludeFromUpgradeFilterHook](./kibana-plugin-core-server.savedobjecttypeexcludefromupgradefilterhook.md)
+
+## SavedObjectTypeExcludeFromUpgradeFilterHook type
+
+If defined, allows a type to run a search query and return a query filter that may match any documents which may be excluded from the next migration upgrade process. Useful for cleaning up large numbers of old documents which are no longer needed and may slow the migration process.
+
+If this hook fails, the migration will proceed without these documents having been filtered out, so this should not be used as a guarantee that these documents have been deleted.
+
+Experimental and subject to change
+
+<b>Signature:</b>
+
+```typescript
+export declare type SavedObjectTypeExcludeFromUpgradeFilterHook = (toolkit: {
+    readonlyEsClient: Pick<ElasticsearchClient, 'search'>;
+}) => estypes.QueryDslQueryContainer | Promise<estypes.QueryDslQueryContainer>;
+```

src/core/server/index.ts

@@ -312,6 +312,7 @@ export type {
   SavedObjectsCreatePointInTimeFinderDependencies,
   SavedObjectsCreatePointInTimeFinderOptions,
   SavedObjectsCreateOptions,
+  SavedObjectTypeExcludeFromUpgradeFilterHook,
   SavedObjectsExportResultDetails,
   SavedObjectsExportExcludedObject,
   SavedObjectsFindResult,

src/core/server/saved_objects/index.ts

@@ -90,6 +90,7 @@ export type {
   SavedObjectStatusMeta,
   SavedObjectsType,
   SavedObjectsTypeManagementDefinition,
+  SavedObjectTypeExcludeFromUpgradeFilterHook,
 } from './types';
 
 export { savedObjectsConfig, savedObjectsMigrationConfig } from './saved_objects_config';

src/core/server/saved_objects/migrationsv2/actions/calculate_exclude_filters.test.ts

@@ -0,0 +1,113 @@
+/*
+ * 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 { errors as esErrors } from '@elastic/elasticsearch';
+import { elasticsearchClientMock } from '../../../elasticsearch/client/mocks';
+import { calculateExcludeFilters } from './calculate_exclude_filters';
+
+describe('calculateExcludeFilters', () => {
+  const client = elasticsearchClientMock.createInternalClient();
+
+  it('calls each provided hook and returns combined filter', async () => {
+    const hook1 = jest.fn().mockReturnValue({ bool: { must: { term: { fieldA: '123' } } } });
+    const hook2 = jest.fn().mockResolvedValue({ bool: { must: { term: { fieldB: 'abc' } } } });
+
+    const task = calculateExcludeFilters({
+      client,
+      excludeFromUpgradeFilterHooks: { type1: hook1, type2: hook2 },
+    });
+    const result = await task();
+
+    expect(hook1).toHaveBeenCalledWith({ readonlyEsClient: { search: expect.any(Function) } });
+    expect(hook2).toHaveBeenCalledWith({ readonlyEsClient: { search: expect.any(Function) } });
+    expect(Either.isRight(result)).toBe(true);
+    expect((result as Either.Right<any>).right).toEqual({
+      excludeFilter: {
+        bool: {
+          must_not: [
+            { bool: { must: { term: { fieldA: '123' } } } },
+            { bool: { must: { term: { fieldB: 'abc' } } } },
+          ],
+        },
+      },
+      errorsByType: {},
+    });
+  });
+
+  it('ignores hooks that return non-retryable errors', async () => {
+    const error = new Error('blah!');
+    const hook1 = jest.fn().mockRejectedValue(error);
+    const hook2 = jest.fn().mockResolvedValue({ bool: { must: { term: { fieldB: 'abc' } } } });
+
+    const task = calculateExcludeFilters({
+      client,
+      excludeFromUpgradeFilterHooks: { type1: hook1, type2: hook2 },
+    });
+    const result = await task();
+
+    expect(Either.isRight(result)).toBe(true);
+    expect((result as Either.Right<any>).right).toEqual({
+      excludeFilter: {
+        bool: {
+          must_not: [{ bool: { must: { term: { fieldB: 'abc' } } } }],
+        },
+      },
+      errorsByType: { type1: error },
+    });
+  });
+
+  it('returns Either.Left if a hook returns a retryable error', async () => {
+    const error = new esErrors.NoLivingConnectionsError(
+      'reason',
+      elasticsearchClientMock.createApiResponse()
+    );
+    const hook1 = jest.fn().mockRejectedValue(error);
+    const hook2 = jest.fn().mockResolvedValue({ bool: { must: { term: { fieldB: 'abc' } } } });
+
+    const task = calculateExcludeFilters({
+      client,
+      excludeFromUpgradeFilterHooks: { type1: hook1, type2: hook2 },
+    });
+    const result = await task();
+
+    expect(Either.isLeft(result)).toBe(true);
+    expect((result as Either.Left<any>).left).toEqual({
+      type: 'retryable_es_client_error',
+      message: 'reason',
+      error,
+    });
+  });
+
+  it('ignores and returns errors for hooks that take longer than timeout', async () => {
+    const hook1 = jest.fn().mockReturnValue(new Promise((r) => setTimeout(r, 40_000)));
+    const hook2 = jest.fn().mockResolvedValue({ bool: { must: { term: { fieldB: 'abc' } } } });
+
+    const task = calculateExcludeFilters({
+      client,
+      excludeFromUpgradeFilterHooks: { type1: hook1, type2: hook2 },
+      hookTimeoutMs: 100,
+    });
+    const resultPromise = task();
+    await new Promise((r) => setTimeout(r, 110));
+    const result = await resultPromise;
+
+    expect(Either.isRight(result)).toBe(true);
+    expect((result as Either.Right<any>).right).toEqual({
+      excludeFilter: {
+        bool: {
+          must_not: [{ bool: { must: { term: { fieldB: 'abc' } } } }],
+        },
+      },
+      errorsByType: expect.any(Object),
+    });
+    expect((result as Either.Right<any>).right.errorsByType.type1.toString()).toMatchInlineSnapshot(
+      `"Error: excludeFromUpgrade hook timed out after 0.1 seconds."`
+    );
+  });
+});

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

@@ -0,0 +1,107 @@
+/*
+ * 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 { estypes } from '@elastic/elasticsearch';
+import { withTimeout } from '@kbn/std';
+import * as Either from 'fp-ts/lib/Either';
+import * as TaskEither from 'fp-ts/lib/TaskEither';
+import { RetryableEsClientError } from '.';
+import { ElasticsearchClient } from '../../../elasticsearch';
+import { SavedObjectTypeExcludeFromUpgradeFilterHook } from '../../types';
+import { catchRetryableEsClientErrors } from './catch_retryable_es_client_errors';
+
+export interface CalculateExcludeFiltersParams {
+  client: ElasticsearchClient;
+  excludeFromUpgradeFilterHooks: Record<string, SavedObjectTypeExcludeFromUpgradeFilterHook>;
+  hookTimeoutMs?: number;
+}
+
+export interface CalculatedExcludeFilter {
+  /** Composite filter of all calculated filters */
+  excludeFilter: estypes.QueryDslQueryContainer;
+  /** Any errors that were encountered during filter calculation, keyed by the type name */
+  errorsByType: Record<string, Error>;
+}
+
+export const calculateExcludeFilters = ({
+  client,
+  excludeFromUpgradeFilterHooks,
+  hookTimeoutMs = 30_000, // default to 30s, exposed for testing
+}: CalculateExcludeFiltersParams): TaskEither.TaskEither<
+  RetryableEsClientError,
+  CalculatedExcludeFilter
+> => () => {
+  return Promise.all<
+    | Either.Right<estypes.QueryDslQueryContainer>
+    | Either.Left<{ soType: string; error: Error | RetryableEsClientError }>
+  >(
+    Object.entries(excludeFromUpgradeFilterHooks).map(([soType, hook]) =>
+      withTimeout({
+        promise: Promise.resolve(
+          hook({ readonlyEsClient: { search: client.search.bind(client) } })
+        ),
+        timeoutMs: hookTimeoutMs,
+      })
+        .then((result) =>
+          result.timedout
+            ? Either.left({
+                soType,
+                error: new Error(
+                  `excludeFromUpgrade hook timed out after ${hookTimeoutMs / 1000} seconds.`
+                ),
+              })
+            : Either.right(result.value)
+        )
+        .catch((error) => {
+          const retryableError = catchRetryableEsClientErrors(error);
+          if (Either.isLeft(retryableError)) {
+            return Either.left({ soType, error: retryableError.left });
+          } else {
+            // Really should never happen, only here to satisfy TypeScript
+            return Either.left({
+              soType,
+              error: new Error(
+                `Unexpected return value from catchRetryableEsClientErrors: "${retryableError.toString()}"`
+              ),
+            });
+          }
+        })
+        .catch((error: Error) => Either.left({ soType, error }))
+    )
+  ).then((results) => {
+    const retryableError = results.find(
+      (r) =>
+        Either.isLeft(r) &&
+        !(r.left.error instanceof Error) &&
+        r.left.error.type === 'retryable_es_client_error'
+    ) as Either.Left<{ soType: string; error: RetryableEsClientError }> | undefined;
+    if (retryableError) {
+      return Either.left(retryableError.left.error);
+    }
+
+    const errorsByType: Array<[string, Error]> = [];
+    const filters: estypes.QueryDslQueryContainer[] = [];
+
+    // Loop through all results and collect successes and errors
+    results.forEach((r) =>
+      Either.isRight(r)
+        ? filters.push(r.right)
+        : Either.isLeft(r) && errorsByType.push([r.left.soType, r.left.error as Error])
+    );
+
+    // Composite filter from all calculated filters that successfully executed
+    const excludeFilter: estypes.QueryDslQueryContainer = {
+      bool: { must_not: filters },
+    };
+
+    return Either.right({
+      excludeFilter,
+      errorsByType: Object.fromEntries(errorsByType),
+    });
+  });
+};

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

@@ -98,6 +98,12 @@ export { searchForOutdatedDocuments } from './search_for_outdated_documents';
 export type { BulkOverwriteTransformedDocumentsParams } from './bulk_overwrite_transformed_documents';
 export { bulkOverwriteTransformedDocuments } from './bulk_overwrite_transformed_documents';
 
+export type {
+  CalculateExcludeFiltersParams,
+  CalculatedExcludeFilter,
+} from './calculate_exclude_filters';
+export { calculateExcludeFilters } from './calculate_exclude_filters';
+
 export { pickupUpdatedMappings, waitForTask, waitForIndexStatusYellow };
 export type { AliasNotFound, RemoveIndexNotAConcreteIndex };
 

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

@@ -39,6 +39,7 @@ describe('createInitialState', () => {
       batchSize: 1000,
       controlState: 'INIT',
       currentAlias: '.kibana_task_manager',
+      excludeFromUpgradeFilterHooks: {},
       indexPrefix: '.kibana_task_manager',
       kibanaVersion: '8.1.0',
       knownTypes: [],
@@ -136,6 +137,31 @@ describe('createInitialState', () => {
     expect(initialState.knownTypes).toEqual(['foo', 'bar']);
   });
 
+  it('returns state with the correct `excludeFromUpgradeFilterHooks`', () => {
+    const fooExcludeOnUpgradeHook = jest.fn();
+    typeRegistry.registerType({
+      name: 'foo',
+      namespaceType: 'single',
+      hidden: false,
+      mappings: { properties: {} },
+      excludeOnUpgrade: fooExcludeOnUpgradeHook,
+    });
+
+    const initialState = createInitialState({
+      kibanaVersion: '8.1.0',
+      targetMappings: {
+        dynamic: 'strict',
+        properties: { my_type: { properties: { title: { type: 'text' } } } },
+      },
+      migrationVersionPerType: {},
+      indexPrefix: '.kibana_task_manager',
+      migrationsConfig,
+      typeRegistry,
+    });
+
+    expect(initialState.excludeFromUpgradeFilterHooks).toEqual({ foo: fooExcludeOnUpgradeHook });
+  });
+
   it('returns state with a preMigration script', () => {
     const preMigrationScript = "ctx._id = ctx._source.type + ':' + ctx._id";
     const initialState = createInitialState({

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

@@ -58,6 +58,12 @@ export const createInitialState = ({
   };
 
   const knownTypes = typeRegistry.getAllTypes().map((type) => type.name);
+  const excludeFilterHooks = Object.fromEntries(
+    typeRegistry
+      .getAllTypes()
+      .filter((type) => !!type.excludeOnUpgrade)
+      .map((type) => [type.name, type.excludeOnUpgrade!])
+  );
 
   return {
     controlState: 'INIT',
@@ -79,5 +85,6 @@ export const createInitialState = ({
     logs: [],
     unusedTypesQuery: excludeUnusedTypesQuery,
     knownTypes,
+    excludeFromUpgradeFilterHooks: excludeFilterHooks,
   };
 };