github-mirrors/kibana
1
0
Fork 0
mirror of https://github.com/elastic/kibana.git synced 2025-04-23 09:19:04 -04:00
Code Issues Projects Releases Packages Wiki Activity

ZDT migration algorithm: allow to switch from a v2-managed state (#158434)

Browse source 
## Summary

Fix https://github.com/elastic/kibana/issues/158372

Change the logic of the `zdt` algorithm to be able to re-use an existing
"v2 managed" cluster state (a kibana cluster previously running the v2
migration algorithm) if compatible.

Technically the v2 index state is compatible with zdt after 8.8 (when we
introduced `indexTypeMap` and restricted to compatible mapping changes
only).

See https://github.com/elastic/kibana/issues/158372 for more detailed
explanations.
This commit is contained in:
Pierre Gayvallet 2023-05-30 04:25:14 -04:00 • committed by GitHub
parent 0d3b63b1c1
commit e12b716ae2
No known key found for this signature in database
GPG key ID: 4AEE18F83AFDEB23
22 changed files with 856 additions and 245 deletions
Download patch file Download diff file Expand all files Collapse all files

1
packages/core/saved-objects/core-saved-objects-migration-server-internal/src/zdt/model/stages/create_target_index.ts
View file

@ -53,5 +53,6 @@ export const createTargetIndex: ModelStage<
aliases: [],
aliasActions,
skipDocumentMigration: true,
previousAlgorithm: 'zdt',
};
};

16
packages/core/saved-objects/core-saved-objects-migration-server-internal/src/zdt/model/stages/documents_update_init.test.ts
View file

@ -46,6 +46,22 @@ describe('Stage: documentsUpdateInit', () => {
context.typeRegistry.registerType(createType({ name: 'bar' }));
});
describe('when state.previousAlgorithm is `v2`', () => {
it('DOCUMENTS_UPDATE_INIT -> SET_DOC_MIGRATION_STARTED', () => {
const state = createState({
previousAlgorithm: 'v2',
});
const res: ResponseType<'DOCUMENTS_UPDATE_INIT'> = Either.right('noop' as const);
const newState = documentsUpdateInit(
state,
res as StateActionResponse<'DOCUMENTS_UPDATE_INIT'>,
context
);
expect(newState.controlState).toEqual('SET_DOC_MIGRATION_STARTED');
});
});
describe('when checkVersionCompatibility returns `greater`', () => {
beforeEach(() => {
checkVersionCompatibilityMock.mockReturnValue({

45
packages/core/saved-objects/core-saved-objects-migration-server-internal/src/zdt/model/stages/documents_update_init.ts
View file

@ -27,6 +27,34 @@ export const documentsUpdateInit: ModelStage<
const types = context.types.map((type) => context.typeRegistry.getType(type)!);
const logs: MigrationLog[] = [...state.logs];
const excludeFilterHooks = Object.fromEntries(
context.types
.map((name) => context.typeRegistry.getType(name)!)
.filter((type) => !!type.excludeOnUpgrade)
.map((type) => [type.name, type.excludeOnUpgrade!])
);
const outdatedDocumentsQuery = getOutdatedDocumentsQuery({ types });
const transformRawDocs = createDocumentTransformFn({
serializer: context.serializer,
documentMigrator: context.documentMigrator,
});
const commonState = {
logs,
excludeOnUpgradeQuery: excludeUnusedTypesQuery,
excludeFromUpgradeFilterHooks: excludeFilterHooks,
outdatedDocumentsQuery,
transformRawDocs,
};
// index was previously using the v2 algo, we skip compat check and jump to next stage
if (state.previousAlgorithm === 'v2') {
return {
...state,
...commonState,
controlState: 'SET_DOC_MIGRATION_STARTED',
};
}
const versionCheck = checkVersionCompatibility({
mappings: state.previousMappings,
types,
@ -43,24 +71,9 @@ export const documentsUpdateInit: ModelStage<
// app version is greater than the index mapping version.
// scenario of an upgrade: we need to run the document migration.
case 'greater':
const excludeFilterHooks = Object.fromEntries(
context.types
.map((name) => context.typeRegistry.getType(name)!)
.filter((type) => !!type.excludeOnUpgrade)
.map((type) => [type.name, type.excludeOnUpgrade!])
);
const outdatedDocumentsQuery = getOutdatedDocumentsQuery({ types });
const transformRawDocs = createDocumentTransformFn({
serializer: context.serializer,
documentMigrator: context.documentMigrator,
});
return {
...state,
logs,
excludeOnUpgradeQuery: excludeUnusedTypesQuery,
excludeFromUpgradeFilterHooks: excludeFilterHooks,
outdatedDocumentsQuery,
transformRawDocs,
...commonState,
controlState: 'SET_DOC_MIGRATION_STARTED',
};
// app version and index mapping version are the same.

12
packages/core/saved-objects/core-saved-objects-migration-server-internal/src/zdt/model/stages/init.test.mocks.ts
View file

@ -11,6 +11,7 @@ export const checkVersionCompatibilityMock = jest.fn();
export const buildIndexMappingsMock = jest.fn();
export const generateAdditiveMappingDiffMock = jest.fn();
export const getAliasActionsMock = jest.fn();
export const checkIndexCurrentAlgorithmMock = jest.fn();
jest.doMock('../../utils', () => {
const realModule = jest.requireActual('../../utils');
@ -21,5 +22,16 @@ jest.doMock('../../utils', () => {
buildIndexMappings: buildIndexMappingsMock,
generateAdditiveMappingDiff: generateAdditiveMappingDiffMock,
getAliasActions: getAliasActionsMock,
checkIndexCurrentAlgorithm: checkIndexCurrentAlgorithmMock,
};
});
export const getAliasesMock = jest.fn();
jest.doMock('../../../model/helpers', () => {
const realModule = jest.requireActual('../../../model/helpers');
return {
...realModule,
getAliases: getAliasesMock,
};
});

484
packages/core/saved-objects/core-saved-objects-migration-server-internal/src/zdt/model/stages/init.test.ts
View file

@ -12,6 +12,8 @@ import {
buildIndexMappingsMock,
generateAdditiveMappingDiffMock,
getAliasActionsMock,
checkIndexCurrentAlgorithmMock,
getAliasesMock,
} from './init.test.mocks';
import * as Either from 'fp-ts/lib/Either';
import { FetchIndexResponse } from '../../../actions';
@ -52,6 +54,9 @@ describe('Stage: init', () => {
});
generateAdditiveMappingDiffMock.mockReset().mockReturnValue({});
getAliasActionsMock.mockReset().mockReturnValue([]);
checkIndexCurrentAlgorithmMock.mockReset().mockReturnValue('zdt');
getAliasesMock.mockReset().mockReturnValue(Either.right({}));
buildIndexMappingsMock.mockReset().mockReturnValue({});
context = createContextMock({ indexPrefix: '.kibana', types: ['foo', 'bar'] });
context.typeRegistry.registerType({
@ -87,10 +92,17 @@ describe('Stage: init', () => {
const fetchIndexResponse = createResponse();
const res: StateActionResponse<'INIT'> = Either.right(fetchIndexResponse);
const aliases = { '.foo': '.bar' };
getAliasesMock.mockReturnValue(Either.right(aliases));
init(state, res, context);
expect(getCurrentIndexMock).toHaveBeenCalledTimes(1);
expect(getCurrentIndexMock).toHaveBeenCalledWith(fetchIndexResponse, context.indexPrefix);
expect(getCurrentIndexMock).toHaveBeenCalledWith({
indices: Object.keys(fetchIndexResponse),
indexPrefix: context.indexPrefix,
aliases,
});
});
it('calls checkVersionCompatibility with the correct parameters', () => {
@ -109,25 +121,100 @@ describe('Stage: init', () => {
});
});
describe('when getCurrentIndex returns undefined', () => {
describe('when checkIndexCurrentAlgorithm returns `unknown`', () => {
beforeEach(() => {
getCurrentIndexMock.mockReturnValue(undefined);
checkIndexCurrentAlgorithmMock.mockReset().mockReturnValue('unknown');
});
it('calls buildIndexMappings with the correct parameters', () => {
it('adds a log entry about the algo check', () => {
const state = createState();
const res: StateActionResponse<'INIT'> = Either.right(createResponse());
const newState = init(state, res, context);
expect(newState.logs.map((entry) => entry.message)).toContain(
`INIT: current algo check result: unknown`
);
});
it('INIT -> FATAL', () => {
const state = createState();
const fetchIndexResponse = createResponse();
const res: StateActionResponse<'INIT'> = Either.right(fetchIndexResponse);
const newState = init(state, res, context);
expect(newState).toEqual(
expect.objectContaining({
controlState: 'FATAL',
reason: 'Cannot identify algorithm used for index .kibana_1',
})
);
});
});
describe('when checkIndexCurrentAlgorithm returns `v2-incompatible`', () => {
beforeEach(() => {
checkIndexCurrentAlgorithmMock.mockReset().mockReturnValue('v2-incompatible');
});
it('adds a log entry about the algo check', () => {
const state = createState();
const res: StateActionResponse<'INIT'> = Either.right(createResponse());
const newState = init(state, res, context);
expect(newState.logs.map((entry) => entry.message)).toContain(
`INIT: current algo check result: v2-incompatible`
);
});
it('INIT -> FATAL', () => {
const state = createState();
const fetchIndexResponse = createResponse();
const res: StateActionResponse<'INIT'> = Either.right(fetchIndexResponse);
const newState = init(state, res, context);
expect(newState).toEqual(
expect.objectContaining({
controlState: 'FATAL',
reason: 'Index .kibana_1 is using an incompatible version of the v2 algorithm',
})
);
});
});
describe('when checkIndexCurrentAlgorithm returns `v2-compatible`', () => {
beforeEach(() => {
checkIndexCurrentAlgorithmMock.mockReset().mockReturnValue('v2-compatible');
buildIndexMappingsMock.mockReturnValue({});
});
it('adds a log entry about the algo check', () => {
const state = createState();
const res: StateActionResponse<'INIT'> = Either.right(createResponse());
init(state, res, context);
expect(buildIndexMappingsMock).toHaveBeenCalledTimes(1);
expect(buildIndexMappingsMock).toHaveBeenCalledWith({
expect(buildIndexMappingsMock).toHaveBeenLastCalledWith({
types: ['foo', 'bar'].map((type) => context.typeRegistry.getType(type)),
});
});
it('INIT -> CREATE_TARGET_INDEX', () => {
it('calls buildIndexMappings with the correct parameters', () => {
const state = createState();
const res: StateActionResponse<'INIT'> = Either.right(createResponse());
const newState = init(state, res, context);
expect(newState.logs.map((entry) => entry.message)).toContain(
`INIT: current algo check result: v2-compatible`
);
});
it('INIT -> UPDATE_INDEX_MAPPINGS', () => {
const state = createState();
const fetchIndexResponse = createResponse();
const res: StateActionResponse<'INIT'> = Either.right(fetchIndexResponse);
@ -137,204 +224,261 @@ describe('Stage: init', () => {
const newState = init(state, res, context);
expect(newState).toEqual(
expect.objectContaining({
controlState: 'CREATE_TARGET_INDEX',
currentIndex: '.kibana_1',
indexMappings: mockMappings,
})
);
});
});
describe('when checkVersionCompatibility returns `greater`', () => {
it('calls generateAdditiveMappingDiff with the correct parameters', () => {
const state = createState();
const fetchIndexResponse = createResponse();
const res: StateActionResponse<'INIT'> = Either.right(fetchIndexResponse);
checkVersionCompatibilityMock.mockReturnValue({
status: 'greater',
});
init(state, res, context);
expect(generateAdditiveMappingDiffMock).toHaveBeenCalledTimes(1);
expect(generateAdditiveMappingDiffMock).toHaveBeenCalledWith({
types: ['foo', 'bar'].map((type) => context.typeRegistry.getType(type)),
meta: fetchIndexResponse[currentIndex].mappings._meta,
deletedTypes: context.deletedTypes,
});
});
it('INIT -> UPDATE_INDEX_MAPPINGS', () => {
const state = createState();
const fetchIndexResponse = createResponse();
const res: StateActionResponse<'INIT'> = Either.right(fetchIndexResponse);
checkVersionCompatibilityMock.mockReturnValue({
status: 'greater',
});
generateAdditiveMappingDiffMock.mockReturnValue({ someToken: {} });
const newState = init(state, res, context);
expect(newState).toEqual(
expect.objectContaining({
controlState: 'UPDATE_INDEX_MAPPINGS',
currentIndex,
previousMappings: fetchIndexResponse[currentIndex].mappings,
additiveMappingChanges: { someToken: {} },
additiveMappingChanges: mockMappings.properties,
skipDocumentMigration: false,
})
);
});
it('adds a log entry about the version check', () => {
const state = createState();
const res: StateActionResponse<'INIT'> = Either.right(createResponse());
checkVersionCompatibilityMock.mockReturnValue({
status: 'greater',
});
const newState = init(state, res, context);
expect(newState.logs.map((entry) => entry.message)).toEqual([
`INIT: mapping version check result: greater`,
]);
});
});
describe('when checkVersionCompatibility returns `equal`', () => {
it('INIT -> UPDATE_ALIASES if alias actions are not empty', () => {
const state = createState();
const fetchIndexResponse = createResponse();
const res: StateActionResponse<'INIT'> = Either.right(fetchIndexResponse);
checkVersionCompatibilityMock.mockReturnValue({
status: 'equal',
});
getAliasActionsMock.mockReturnValue([{ add: { index: '.kibana_1', alias: '.kibana' } }]);
const newState = init(state, res, context);
expect(newState).toEqual(
expect.objectContaining({
controlState: 'UPDATE_ALIASES',
currentIndex,
previousMappings: fetchIndexResponse[currentIndex].mappings,
skipDocumentMigration: false,
})
);
describe('when checkIndexCurrentAlgorithm returns `zdt`', () => {
beforeEach(() => {
checkIndexCurrentAlgorithmMock.mockReset().mockReturnValue('zdt');
});
it('INIT -> INDEX_STATE_UPDATE_DONE if alias actions are empty', () => {
const state = createState();
const fetchIndexResponse = createResponse();
const res: StateActionResponse<'INIT'> = Either.right(fetchIndexResponse);
checkVersionCompatibilityMock.mockReturnValue({
status: 'equal',
});
getAliasActionsMock.mockReturnValue([]);
const newState = init(state, res, context);
expect(newState).toEqual(
expect.objectContaining({
controlState: 'INDEX_STATE_UPDATE_DONE',
currentIndex,
previousMappings: fetchIndexResponse[currentIndex].mappings,
skipDocumentMigration: false,
})
);
});
it('adds a log entry about the version check', () => {
it('adds a log entry about the algo check', () => {
const state = createState();
const res: StateActionResponse<'INIT'> = Either.right(createResponse());
checkVersionCompatibilityMock.mockReturnValue({
status: 'equal',
});
const newState = init(state, res, context);
expect(newState.logs.map((entry) => entry.message)).toEqual([
`INIT: mapping version check result: equal`,
]);
});
});
describe('when checkVersionCompatibility returns `lesser`', () => {
it('INIT -> INDEX_STATE_UPDATE_DONE', () => {
const state = createState();
const fetchIndexResponse = createResponse();
const res: StateActionResponse<'INIT'> = Either.right(fetchIndexResponse);
checkVersionCompatibilityMock.mockReturnValue({
status: 'lesser',
});
const newState = init(state, res, context);
expect(newState).toEqual(
expect.objectContaining({
controlState: 'INDEX_STATE_UPDATE_DONE',
})
expect(newState.logs.map((entry) => entry.message)).toContain(
`INIT: current algo check result: zdt`
);
});
it('adds a log entry about the version check', () => {
const state = createState();
const res: StateActionResponse<'INIT'> = Either.right(createResponse());
checkVersionCompatibilityMock.mockReturnValue({
status: 'lesser',
describe('when getCurrentIndex returns undefined', () => {
beforeEach(() => {
getCurrentIndexMock.mockReturnValue(undefined);
});
const newState = init(state, res, context);
it('calls buildIndexMappings with the correct parameters', () => {
const state = createState();
const fetchIndexResponse = createResponse();
const res: StateActionResponse<'INIT'> = Either.right(fetchIndexResponse);
expect(newState.logs.map((entry) => entry.message)).toEqual([
`INIT: mapping version check result: lesser`,
]);
});
});
init(state, res, context);
describe('when checkVersionCompatibility returns `conflict`', () => {
it('INIT -> FATAL', () => {
const state = createState();
const fetchIndexResponse = createResponse();
const res: StateActionResponse<'INIT'> = Either.right(fetchIndexResponse);
checkVersionCompatibilityMock.mockReturnValue({
status: 'conflict',
expect(buildIndexMappingsMock).toHaveBeenCalledTimes(1);
expect(buildIndexMappingsMock).toHaveBeenCalledWith({
types: ['foo', 'bar'].map((type) => context.typeRegistry.getType(type)),
});
});
const newState = init(state, res, context);
it('INIT -> CREATE_TARGET_INDEX', () => {
const state = createState();
const fetchIndexResponse = createResponse();
const res: StateActionResponse<'INIT'> = Either.right(fetchIndexResponse);
expect(newState).toEqual(
expect.objectContaining({
controlState: 'FATAL',
reason: 'Model version conflict: inconsistent higher/lower versions',
})
);
const mockMappings = { properties: { someMappings: 'string' } };
buildIndexMappingsMock.mockReturnValue(mockMappings);
const newState = init(state, res, context);
expect(newState).toEqual(
expect.objectContaining({
controlState: 'CREATE_TARGET_INDEX',
currentIndex: '.kibana_1',
indexMappings: mockMappings,
})
);
});
});
it('adds a log entry about the version check', () => {
const state = createState();
const res: StateActionResponse<'INIT'> = Either.right(createResponse());
describe('when checkVersionCompatibility returns `greater`', () => {
it('calls generateAdditiveMappingDiff with the correct parameters', () => {
const state = createState();
const fetchIndexResponse = createResponse();
const res: StateActionResponse<'INIT'> = Either.right(fetchIndexResponse);
checkVersionCompatibilityMock.mockReturnValue({
status: 'conflict',
checkVersionCompatibilityMock.mockReturnValue({
status: 'greater',
});
init(state, res, context);
expect(generateAdditiveMappingDiffMock).toHaveBeenCalledTimes(1);
expect(generateAdditiveMappingDiffMock).toHaveBeenCalledWith({
types: ['foo', 'bar'].map((type) => context.typeRegistry.getType(type)),
meta: fetchIndexResponse[currentIndex].mappings._meta,
deletedTypes: context.deletedTypes,
});
});
const newState = init(state, res, context);
it('INIT -> UPDATE_INDEX_MAPPINGS', () => {
const state = createState();
const fetchIndexResponse = createResponse();
const res: StateActionResponse<'INIT'> = Either.right(fetchIndexResponse);
expect(newState.logs.map((entry) => entry.message)).toEqual([
`INIT: mapping version check result: conflict`,
]);
checkVersionCompatibilityMock.mockReturnValue({
status: 'greater',
});
generateAdditiveMappingDiffMock.mockReturnValue({ someToken: {} });
const newState = init(state, res, context);
expect(newState).toEqual(
expect.objectContaining({
controlState: 'UPDATE_INDEX_MAPPINGS',
currentIndex,
previousMappings: fetchIndexResponse[currentIndex].mappings,
additiveMappingChanges: { someToken: {} },
skipDocumentMigration: false,
})
);
});
it('adds a log entry about the version check', () => {
const state = createState();
const res: StateActionResponse<'INIT'> = Either.right(createResponse());
checkVersionCompatibilityMock.mockReturnValue({
status: 'greater',
});
const newState = init(state, res, context);
expect(newState.logs.map((entry) => entry.message)).toContain(
`INIT: mapping version check result: greater`
);
});
});
describe('when checkVersionCompatibility returns `equal`', () => {
it('INIT -> UPDATE_ALIASES if alias actions are not empty', () => {
const state = createState();
const fetchIndexResponse = createResponse();
const res: StateActionResponse<'INIT'> = Either.right(fetchIndexResponse);
checkVersionCompatibilityMock.mockReturnValue({
status: 'equal',
});
getAliasActionsMock.mockReturnValue([{ add: { index: '.kibana_1', alias: '.kibana' } }]);
const newState = init(state, res, context);
expect(newState).toEqual(
expect.objectContaining({
controlState: 'UPDATE_ALIASES',
currentIndex,
previousMappings: fetchIndexResponse[currentIndex].mappings,
skipDocumentMigration: false,
})
);
});
it('INIT -> INDEX_STATE_UPDATE_DONE if alias actions are empty', () => {
const state = createState();
const fetchIndexResponse = createResponse();
const res: StateActionResponse<'INIT'> = Either.right(fetchIndexResponse);
checkVersionCompatibilityMock.mockReturnValue({
status: 'equal',
});
getAliasActionsMock.mockReturnValue([]);
const newState = init(state, res, context);
expect(newState).toEqual(
expect.objectContaining({
controlState: 'INDEX_STATE_UPDATE_DONE',
currentIndex,
previousMappings: fetchIndexResponse[currentIndex].mappings,
skipDocumentMigration: false,
})
);
});
it('adds a log entry about the version check', () => {
const state = createState();
const res: StateActionResponse<'INIT'> = Either.right(createResponse());
checkVersionCompatibilityMock.mockReturnValue({
status: 'equal',
});
const newState = init(state, res, context);
expect(newState.logs.map((entry) => entry.message)).toContain(
`INIT: mapping version check result: equal`
);
});
});
describe('when checkVersionCompatibility returns `lesser`', () => {
it('INIT -> INDEX_STATE_UPDATE_DONE', () => {
const state = createState();
const fetchIndexResponse = createResponse();
const res: StateActionResponse<'INIT'> = Either.right(fetchIndexResponse);
checkVersionCompatibilityMock.mockReturnValue({
status: 'lesser',
});
const newState = init(state, res, context);
expect(newState).toEqual(
expect.objectContaining({
controlState: 'INDEX_STATE_UPDATE_DONE',
})
);
});
it('adds a log entry about the version check', () => {
const state = createState();
const res: StateActionResponse<'INIT'> = Either.right(createResponse());
checkVersionCompatibilityMock.mockReturnValue({
status: 'lesser',
});
const newState = init(state, res, context);
expect(newState.logs.map((entry) => entry.message)).toContain(
`INIT: mapping version check result: lesser`
);
});
});
describe('when checkVersionCompatibility returns `conflict`', () => {
it('INIT -> FATAL', () => {
const state = createState();
const fetchIndexResponse = createResponse();
const res: StateActionResponse<'INIT'> = Either.right(fetchIndexResponse);
checkVersionCompatibilityMock.mockReturnValue({
status: 'conflict',
});
const newState = init(state, res, context);
expect(newState).toEqual(
expect.objectContaining({
controlState: 'FATAL',
reason: 'Model version conflict: inconsistent higher/lower versions',
})
);
});
it('adds a log entry about the version check', () => {
const state = createState();
const res: StateActionResponse<'INIT'> = Either.right(createResponse());
checkVersionCompatibilityMock.mockReturnValue({
status: 'conflict',
});
const newState = init(state, res, context);
expect(newState.logs.map((entry) => entry.message)).toContain(
`INIT: mapping version check result: conflict`
);
});
});
});
});

106
packages/core/saved-objects/core-saved-objects-migration-server-internal/src/zdt/model/stages/init.ts
View file

@ -12,12 +12,15 @@ import { delayRetryState } from '../../../model/retry_state';
import { throwBadResponse } from '../../../model/helpers';
import type { MigrationLog } from '../../../types';
import { isTypeof } from '../../actions';
import { getAliases } from '../../../model/helpers';
import {
getCurrentIndex,
checkVersionCompatibility,
buildIndexMappings,
getAliasActions,
generateAdditiveMappingDiff,
checkIndexCurrentAlgorithm,
removePropertiesFromV2,
} from '../../utils';
import type { ModelStage } from '../types';
@ -43,9 +46,25 @@ export const init: ModelStage<
const logs: MigrationLog[] = [...state.logs];
const indices = res.right;
const currentIndex = getCurrentIndex(indices, context.indexPrefix);
const aliasesRes = getAliases(indices);
if (Either.isLeft(aliasesRes)) {
return {
...state,
controlState: 'FATAL',
reason: `The ${
aliasesRes.left.alias
} alias is pointing to multiple indices: ${aliasesRes.left.indices.join(',')}.`,
};
}
const aliasMap = aliasesRes.right;
// No indices were found, likely because it is the first time Kibana boots.
const currentIndex = getCurrentIndex({
indices: Object.keys(indices),
aliases: aliasMap,
indexPrefix: context.indexPrefix,
});
// No indices were found, likely because it is a fresh cluster.
// In that case, we just create the index.
if (!currentIndex) {
return {
@ -60,6 +79,70 @@ export const init: ModelStage<
// Index was found. This is the standard scenario, we check the model versions
// compatibility before going further.
const currentMappings = indices[currentIndex].mappings;
// Index is already present, so we check which algo was last used on it
const currentAlgo = checkIndexCurrentAlgorithm(currentMappings);
logs.push({
level: 'info',
message: `INIT: current algo check result: ${currentAlgo}`,
});
// incompatible (pre 8.8/index-split https://github.com/elastic/kibana/pull/154888) v2 algo => we terminate
if (currentAlgo === 'v2-incompatible') {
return {
...state,
logs,
controlState: 'FATAL',
reason: `Index ${currentIndex} is using an incompatible version of the v2 algorithm`,
};
}
// unknown algo => we terminate
if (currentAlgo === 'unknown') {
return {
...state,
logs,
controlState: 'FATAL',
reason: `Cannot identify algorithm used for index ${currentIndex}`,
};
}
const existingAliases = Object.keys(indices[currentIndex].aliases);
const aliasActions = getAliasActions({
existingAliases,
currentIndex,
indexPrefix: context.indexPrefix,
kibanaVersion: context.kibanaVersion,
});
// cloning as we may be mutating it in later stages.
let currentIndexMeta = cloneDeep(currentMappings._meta!);
if (currentAlgo === 'v2-compatible') {
currentIndexMeta = removePropertiesFromV2(currentIndexMeta);
}
const commonState = {
logs,
currentIndex,
currentIndexMeta,
aliases: existingAliases,
aliasActions,
previousMappings: currentMappings,
previousAlgorithm: currentAlgo === 'v2-compatible' ? ('v2' as const) : ('zdt' as const),
};
// compatible (8.8+) v2 algo => we jump to update index mapping
if (currentAlgo === 'v2-compatible') {
const indexMappings = buildIndexMappings({ types });
return {
...state,
controlState: 'UPDATE_INDEX_MAPPINGS',
...commonState,
additiveMappingChanges: indexMappings.properties,
};
}
// Index was found and is already using ZDT algo. This is the standard scenario.
// We check the model versions compatibility before going further.
const versionCheck = checkVersionCompatibility({
mappings: currentMappings,
types,
@ -72,25 +155,6 @@ export const init: ModelStage<
message: `INIT: mapping version check result: ${versionCheck.status}`,
});
const aliases = Object.keys(indices[currentIndex].aliases);
const aliasActions = getAliasActions({
existingAliases: aliases,
currentIndex,
indexPrefix: context.indexPrefix,
kibanaVersion: context.kibanaVersion,
});
// cloning as we may be mutating it in later stages.
const currentIndexMeta = cloneDeep(currentMappings._meta!);
const commonState = {
logs,
currentIndex,
currentIndexMeta,
aliases,
aliasActions,
previousMappings: currentMappings,
};
switch (versionCheck.status) {
// app version is greater than the index mapping version.
// scenario of an upgrade: we need to update the mappings

7
packages/core/saved-objects/core-saved-objects-migration-server-internal/src/zdt/state/types.ts
View file

@ -17,7 +17,7 @@ import type { MigrationLog, Progress, TransformRawDocs } from '../../types';
import type { ControlState } from '../../state_action_machine';
import type { BulkOperationBatch } from '../../model/create_batches';
import type { AliasAction } from '../../actions';
import { TransformErrorObjects } from '../../core';
import type { TransformErrorObjects } from '../../core';
export interface BaseState extends ControlState {
readonly retryCount: number;
@ -65,6 +65,11 @@ export interface PostInitState extends BaseState {
* All operations updating this field will update in the state accordingly.
*/
readonly currentIndexMeta: IndexMappingMeta;
/**
* The previous algorithm that was last used to migrate this index.
* Used for v2->zdt state conversion.
*/
readonly previousAlgorithm: 'zdt' | 'v2';
}
/**

1
packages/core/saved-objects/core-saved-objects-migration-server-internal/src/zdt/test_helpers/state.ts
View file

@ -19,6 +19,7 @@ export const createPostInitState = (): PostInitState => ({
previousMappings: { properties: {} },
currentIndexMeta: {},
skipDocumentMigration: false,
previousAlgorithm: 'zdt',
});
export const createPostDocInitState = (): PostDocInitState => ({

81
packages/core/saved-objects/core-saved-objects-migration-server-internal/src/zdt/utils/check_index_algorithm.test.ts Normal file
View file

@ -0,0 +1,81 @@
/*
* 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 type { IndexMapping } from '@kbn/core-saved-objects-base-server-internal';
import { checkIndexCurrentAlgorithm } from './check_index_algorithm';
describe('checkIndexCurrentAlgorithm', () => {
it('returns `unknown` if _meta is not present on the mapping', () => {
const mapping: IndexMapping = {
properties: {},
};
expect(checkIndexCurrentAlgorithm(mapping)).toEqual('unknown');
});
it('returns `unknown` if both v2 and zdt metas are present', () => {
const mapping: IndexMapping = {
properties: {},
_meta: {
migrationMappingPropertyHashes: {
foo: 'someHash',
},
docVersions: {
foo: '8.8.0',
},
},
};
expect(checkIndexCurrentAlgorithm(mapping)).toEqual('unknown');
});
it('returns `zdt` if only zdt metas are present', () => {
const mapping: IndexMapping = {
properties: {},
_meta: {
docVersions: {
foo: '8.8.0',
},
mappingVersions: {
foo: '8.8.0',
},
},
};
expect(checkIndexCurrentAlgorithm(mapping)).toEqual('zdt');
});
it('returns `v2-incompatible` if v2 hashes are present but not indexTypesMap', () => {
const mapping: IndexMapping = {
properties: {},
_meta: {
migrationMappingPropertyHashes: {
foo: 'someHash',
},
},
};
expect(checkIndexCurrentAlgorithm(mapping)).toEqual('v2-incompatible');
});
it('returns `v2-compatible` if v2 hashes and indexTypesMap are present', () => {
const mapping: IndexMapping = {
properties: {},
_meta: {
migrationMappingPropertyHashes: {
foo: 'someHash',
},
indexTypesMap: {
'.kibana': ['foo'],
},
},
};
expect(checkIndexCurrentAlgorithm(mapping)).toEqual('v2-compatible');
});
});

43
packages/core/saved-objects/core-saved-objects-migration-server-internal/src/zdt/utils/check_index_algorithm.ts Normal file
View file

@ -0,0 +1,43 @@
/*
* 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 type { IndexMapping } from '@kbn/core-saved-objects-base-server-internal';
/**
* The list of values returned by `checkIndexCurrentAlgorithm`.
*
* - `zdt`
* - `v2-compatible`
* - `v2-incompatible`
* - `unknown`
*/
export type CheckCurrentAlgorithmResult = 'zdt' | 'v2-compatible' | 'v2-incompatible' | 'unknown';
export const checkIndexCurrentAlgorithm = (
indexMapping: IndexMapping
): CheckCurrentAlgorithmResult => {
const meta = indexMapping._meta;
if (!meta) {
return 'unknown';
}
const hasV2Meta = !!meta.migrationMappingPropertyHashes;
const hasZDTMeta = !!meta.docVersions || !!meta.mappingVersions;
if (hasV2Meta && hasZDTMeta) {
return 'unknown';
}
if (hasV2Meta) {
const isCompatible = !!meta.indexTypesMap;
return isCompatible ? 'v2-compatible' : 'v2-incompatible';
}
if (hasZDTMeta) {
return 'zdt';
}
return 'unknown';
};

79
packages/core/saved-objects/core-saved-objects-migration-server-internal/src/zdt/utils/get_current_index.test.ts
View file

@ -7,42 +7,79 @@
*/
import { getCurrentIndex } from './get_current_index';
import type { FetchIndexResponse } from '../../actions';
describe('getCurrentIndex', () => {
const createIndexResponse = (...indexNames: string[]): FetchIndexResponse => {
return indexNames.reduce<FetchIndexResponse>((resp, indexName) => {
resp[indexName] = { aliases: {}, mappings: { properties: {} }, settings: {} };
return resp;
}, {});
};
it('returns the target of the alias matching the index prefix if present in the list', () => {
expect(
getCurrentIndex({
indices: ['.kibana_1', '.kibana_2', '.kibana_8.8.0_001'],
indexPrefix: '.kibana',
aliases: {
'.kibana': '.kibana_8.8.0_001',
},
})
).toEqual('.kibana_8.8.0_001');
});
it('ignores the target of the alias matching the index prefix if not present in the list', () => {
expect(
getCurrentIndex({
indices: ['.kibana_1'],
indexPrefix: '.kibana',
aliases: {
'.kibana': '.foobar_9000',
},
})
).toEqual('.kibana_1');
});
it('returns the highest numbered index matching the index prefix', () => {
const resp = createIndexResponse('.kibana_1', '.kibana_2');
expect(getCurrentIndex(resp, '.kibana')).toEqual('.kibana_2');
expect(
getCurrentIndex({
indices: ['.kibana_1', '.kibana_2'],
indexPrefix: '.kibana',
aliases: {},
})
).toEqual('.kibana_2');
});
it('ignores other indices', () => {
const resp = createIndexResponse('.kibana_1', '.kibana_2', '.foo_3');
expect(getCurrentIndex(resp, '.kibana')).toEqual('.kibana_2');
expect(
getCurrentIndex({
indices: ['.kibana_1', '.kibana_2', '.foo_3'],
indexPrefix: '.kibana',
aliases: {},
})
).toEqual('.kibana_2');
});
it('ignores other indices including the prefix', () => {
const resp = createIndexResponse('.kibana_1', '.kibana_2', '.kibana_task_manager_3');
expect(getCurrentIndex(resp, '.kibana')).toEqual('.kibana_2');
expect(
getCurrentIndex({
indices: ['.kibana_1', '.kibana_2', '.kibana_task_manager_3'],
indexPrefix: '.kibana',
aliases: {},
})
).toEqual('.kibana_2');
});
it('ignores other indices including a subpart of the prefix', () => {
const resp = createIndexResponse(
'.kibana_3',
'.kibana_task_manager_1',
'.kibana_task_manager_2'
);
expect(getCurrentIndex(resp, '.kibana_task_manager')).toEqual('.kibana_task_manager_2');
expect(
getCurrentIndex({
indices: ['.kibana_3', '.kibana_task_manager_1', '.kibana_task_manager_2'],
indexPrefix: '.kibana_task_manager',
aliases: {},
})
).toEqual('.kibana_task_manager_2');
});
it('returns undefined if no indices match', () => {
const resp = createIndexResponse('.kibana_task_manager_1', '.kibana_task_manager_2');
expect(getCurrentIndex(resp, '.kibana')).toBeUndefined();
expect(
getCurrentIndex({
indices: ['.kibana_task_manager_1', '.kibana_task_manager_2'],
indexPrefix: '.kibana',
aliases: {},
})
).toBeUndefined();
});
});

22
packages/core/saved-objects/core-saved-objects-migration-server-internal/src/zdt/utils/get_current_index.ts
View file

@ -7,16 +7,26 @@
*/
import { escapeRegExp } from 'lodash';
import type { FetchIndexResponse } from '../../actions';
import type { Aliases } from '../../model/helpers';
export const getCurrentIndex = ({
indices,
aliases,
indexPrefix,
}: {
indices: string[];
aliases: Aliases;
indexPrefix: string;
}): string | undefined => {
// if there is already a current alias pointing to an index, we reuse the index.
if (aliases[indexPrefix] && indices.includes(aliases[indexPrefix]!)) {
return aliases[indexPrefix];
}
export const getCurrentIndex = (
indices: FetchIndexResponse,
indexPrefix: string
): string | undefined => {
const matcher = new RegExp(`^${escapeRegExp(indexPrefix)}[_](?<counter>\\d+)$`);
let lastCount = -1;
Object.keys(indices).forEach((indexName) => {
indices.forEach((indexName) => {
const match = matcher.exec(indexName);
if (match && match.groups?.counter) {
const suffix = parseInt(match.groups.counter, 10);

5
packages/core/saved-objects/core-saved-objects-migration-server-internal/src/zdt/utils/index.ts
View file

@ -18,4 +18,9 @@ export {
setMetaMappingMigrationComplete,
setMetaDocMigrationStarted,
setMetaDocMigrationComplete,
removePropertiesFromV2,
} from './update_index_meta';
export {
checkIndexCurrentAlgorithm,
type CheckCurrentAlgorithmResult,
} from './check_index_algorithm';

18
packages/core/saved-objects/core-saved-objects-migration-server-internal/src/zdt/utils/update_index_meta.test.ts
View file

@ -14,6 +14,7 @@ import {
setMetaDocMigrationStarted,
setMetaDocMigrationComplete,
setMetaMappingMigrationComplete,
removePropertiesFromV2,
} from './update_index_meta';
const getDefaultMeta = (): IndexMappingMeta => ({
@ -80,3 +81,20 @@ describe('setMetaDocMigrationComplete', () => {
});
});
});
describe('removePropertiesFromV2', () => {
it('removes meta properties used by the v2 algorithm', () => {
const meta: IndexMappingMeta = {
...getDefaultMeta(),
indexTypesMap: {
'.kibana': ['foo'],
},
migrationMappingPropertyHashes: {
foo: 'someHash',
},
};
const output = removePropertiesFromV2(meta);
expect(output).toEqual(getDefaultMeta());
});
});

7
packages/core/saved-objects/core-saved-objects-migration-server-internal/src/zdt/utils/update_index_meta.ts
View file

@ -56,3 +56,10 @@ export const setMetaDocMigrationComplete = ({
},
};
};
export const removePropertiesFromV2 = (meta: IndexMappingMeta): IndexMappingMeta => {
const cleaned = { ...meta };
delete cleaned.indexTypesMap;
delete cleaned.migrationMappingPropertyHashes;
return cleaned;
};

10
src/core/server/integration_tests/saved_objects/migrations/fixtures/zdt_base.fixtures.ts
View file

@ -8,19 +8,23 @@
import { SavedObjectsModelVersion, SavedObjectMigrationFn } from '@kbn/core-saved-objects-server';
import { createType } from '../test_utils';
import { type KibanaMigratorTestKitParams } from '../kibana_migrator_test_kit';
import { type KibanaMigratorTestKitParams, currentVersion } from '../kibana_migrator_test_kit';
export const getBaseMigratorParams = ({
migrationAlgorithm = 'zdt',
// default to true here as most tests need to run the full migration
runOnNonMigratorNodes = true,
kibanaVersion = currentVersion,
}: {
runOnNonMigratorNodes?: boolean;
migrationAlgorithm?: 'v2' | 'zdt';
kibanaVersion?: string;
} = {}): KibanaMigratorTestKitParams => ({
kibanaIndex: '.kibana',
kibanaVersion: '8.8.0',
kibanaVersion,
settings: {
migrations: {
algorithm: 'zdt',
algorithm: migrationAlgorithm,
zdt: {
metaPickupSyncDelaySec: 5,
runOnNonMigratorNodes,

2
src/core/server/integration_tests/saved_objects/migrations/zdt_1/create_index.test.ts
View file

@ -34,7 +34,7 @@ describe('ZDT upgrades - running on a fresh cluster', () => {
const barType = getBarType();
const { runMigrations, client } = await getKibanaMigratorTestKit({
...getBaseMigratorParams(),
...getBaseMigratorParams({ kibanaVersion: '8.8.0' }),
logFilePath,
types: [fooType, barType],
});

2
src/core/server/integration_tests/saved_objects/migrations/zdt_1/mapping_version_conflict.test.ts
View file

@ -24,7 +24,7 @@ export const logFilePath = Path.join(__dirname, 'mapping_version_conflict.test.l
describe('ZDT upgrades - mapping model version conflict', () => {
let esServer: TestElasticsearchUtils['es'];
const baseMigratorParams = getBaseMigratorParams();
const baseMigratorParams = getBaseMigratorParams({ kibanaVersion: '8.8.0' });
beforeAll(async () => {
await fs.unlink(logFilePath).catch(() => {});

4
src/core/server/integration_tests/saved_objects/migrations/zdt_1/update_mappings.test.ts
View file

@ -38,7 +38,7 @@ describe('ZDT upgrades - basic mapping update', () => {
const fooType = getFooType();
const barType = getBarType();
const { runMigrations } = await getKibanaMigratorTestKit({
...getBaseMigratorParams(),
...getBaseMigratorParams({ kibanaVersion: '8.8.0' }),
types: [fooType, barType],
});
await runMigrations();
@ -70,7 +70,7 @@ describe('ZDT upgrades - basic mapping update', () => {
};
const { runMigrations, client } = await getKibanaMigratorTestKit({
...getBaseMigratorParams(),
...getBaseMigratorParams({ kibanaVersion: '8.8.0' }),
logFilePath,
types: [fooType, barType],
});

150
src/core/server/integration_tests/saved_objects/migrations/zdt_2/v2_to_zdt_switch.test.ts Normal file
View file

@ -0,0 +1,150 @@
/*
* 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 Path from 'path';
import fs from 'fs/promises';
import { range } from 'lodash';
import { type TestElasticsearchUtils } from '@kbn/core-test-helpers-kbn-server';
import { SavedObjectsBulkCreateObject } from '@kbn/core-saved-objects-api-server';
import { IndexMappingMeta } from '@kbn/core-saved-objects-base-server-internal';
import '../jest_matchers';
import {
getKibanaMigratorTestKit,
startElasticsearch,
currentVersion,
} from '../kibana_migrator_test_kit';
import { delay, parseLogFile } from '../test_utils';
import { getBaseMigratorParams, getSampleAType } from '../fixtures/zdt_base.fixtures';
export const logFilePath = Path.join(__dirname, 'v2_to_zdt_switch.test.log');
const v2IndexName = `.kibana_${currentVersion}_001`;
describe('ZDT upgrades - switching from v2 algorithm', () => {
let esServer: TestElasticsearchUtils['es'];
beforeEach(async () => {
await fs.unlink(logFilePath).catch(() => {});
esServer = await startElasticsearch();
});
afterEach(async () => {
await esServer?.stop();
await delay(10);
});
const createBaseline = async ({
kibanaVersion = currentVersion,
}: { kibanaVersion?: string } = {}) => {
const { runMigrations, savedObjectsRepository, client } = await getKibanaMigratorTestKit({
...getBaseMigratorParams({ migrationAlgorithm: 'v2', kibanaVersion }),
types: [getSampleAType()],
});
await runMigrations();
const sampleAObjs = range(5).map<SavedObjectsBulkCreateObject>((number) => ({
id: `a-${number}`,
type: 'sample_a',
attributes: {
keyword: `a_${number}`,
boolean: true,
},
}));
await savedObjectsRepository.bulkCreate(sampleAObjs);
return { client };
};
describe('when switching from a compatible version', () => {
it('it able to re-use a cluster state from the v2 algorithm', async () => {
await createBaseline();
const typeA = getSampleAType();
const { runMigrations, client, savedObjectsRepository } = await getKibanaMigratorTestKit({
...getBaseMigratorParams({ migrationAlgorithm: 'zdt' }),
logFilePath,
types: [typeA],
});
await runMigrations();
const indices = await client.indices.get({ index: '.kibana*' });
expect(Object.keys(indices)).toEqual([v2IndexName]);
const index = indices[v2IndexName];
const mappings = index.mappings ?? {};
const mappingMeta = mappings._meta ?? {};
expect(mappings.properties).toEqual(
expect.objectContaining({
sample_a: typeA.mappings,
})
);
expect(mappingMeta).toEqual({
docVersions: { sample_a: '10.1.0' },
mappingVersions: { sample_a: '10.1.0' },
migrationState: expect.any(Object),
});
const { saved_objects: sampleADocs } = await savedObjectsRepository.find({
type: 'sample_a',
});
expect(sampleADocs).toHaveLength(5);
const records = await parseLogFile(logFilePath);
expect(records).toContainLogEntries(
[
'INIT: current algo check result: v2-compatible',
'INIT -> UPDATE_INDEX_MAPPINGS',
'INDEX_STATE_UPDATE_DONE -> DOCUMENTS_UPDATE_INIT',
'Migration completed',
],
{ ordered: true }
);
});
});
describe('when switching from an incompatible version', () => {
it('fails and throws an explicit error', async () => {
const { client } = await createBaseline({ kibanaVersion: '8.7.0' });
// even when specifying an older version, the `indexTypeMap` will be present on the index's meta,
// so we have to manually remove it there.
const indices = await client.indices.get({
index: '.kibana_8.7.0_001',
});
const meta = indices['.kibana_8.7.0_001'].mappings!._meta! as IndexMappingMeta;
delete meta.indexTypesMap;
await client.indices.putMapping({
index: '.kibana_8.7.0_001',
_meta: meta,
});
const typeA = getSampleAType();
const { runMigrations } = await getKibanaMigratorTestKit({
...getBaseMigratorParams({ migrationAlgorithm: 'zdt' }),
logFilePath,
types: [typeA],
});
await expect(runMigrations()).rejects.toThrowErrorMatchingInlineSnapshot(
`"Unable to complete saved object migrations for the [.kibana] index: Index .kibana_8.7.0_001 is using an incompatible version of the v2 algorithm"`
);
const records = await parseLogFile(logFilePath);
expect(records).toContainLogEntries(
['current algo check result: v2-incompatible', 'INIT -> FATAL'],
{ ordered: true }
);
});
});
});

2
src/core/server/integration_tests/saved_objects/migrations/zdt_v2_compat/create_index.test.ts
View file

@ -34,7 +34,7 @@ describe('ZDT with v2 compat - running on a fresh cluster', () => {
const legacyType = getLegacyType();
const { runMigrations, client } = await getKibanaMigratorTestKit({
...getBaseMigratorParams(),
...getBaseMigratorParams({ kibanaVersion: '8.8.0' }),
logFilePath,
types: [fooType, legacyType],
});

4
src/core/server/integration_tests/saved_objects/migrations/zdt_v2_compat/update_mappings.test.ts
View file

@ -39,7 +39,7 @@ describe('ZDT with v2 compat - basic mapping update', () => {
const fooType = getFooType();
const legacyType = getLegacyType();
const { runMigrations } = await getKibanaMigratorTestKit({
...getBaseMigratorParams(),
...getBaseMigratorParams({ kibanaVersion: '8.8.0' }),
types: [fooType, legacyType],
});
await runMigrations();
@ -71,7 +71,7 @@ describe('ZDT with v2 compat - basic mapping update', () => {
};
const { runMigrations, client } = await getKibanaMigratorTestKit({
...getBaseMigratorParams(),
...getBaseMigratorParams({ kibanaVersion: '8.8.0' }),
logFilePath,
types: [fooType, legacyType],
});