[saved objects] Add migrations v2 integration test for scenario with …

…multiple Kibana nodes.

src/core/server/saved_objects/migrationsv2/integration_tests/multiple_kibana_nodes.test.ts

@@ -0,0 +1,219 @@
+/*
+ * 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';
+import Util from 'util';
+import glob from 'glob';
+import { esTestConfig, kibanaServerTestUser } from '@kbn/test';
+import { kibanaPackageJson as pkg } from '@kbn/utils';
+import * as kbnTestServer from '../../../../test_helpers/kbn_server';
+import type { ElasticsearchClient } from '../../../elasticsearch';
+import { SavedObjectsType } from '../../types';
+import type { Root } from '../../../root';
+
+const LOG_FILE_PREFIX = 'migration_test_multiple_kibana_nodes';
+
+const asyncUnlink = Util.promisify(Fs.unlink);
+
+async function removeLogFiles() {
+  glob(Path.join(__dirname, `${LOG_FILE_PREFIX}_*.log`), (err, files) => {
+    files.forEach(async (file) => {
+      // ignore errors if it doesn't exist
+      await asyncUnlink(file).catch(() => void 0);
+    });
+  });
+}
+
+function extractSortNumberFromId(id: string): number {
+  const parsedId = parseInt(id.split(':')[1], 10); // "foo:123" -> 123
+  if (isNaN(parsedId)) {
+    throw new Error(`Failed to parse Saved Object ID [${id}]. Result is NaN`);
+  }
+  return parsedId;
+}
+
+async function fetchDocs(esClient: ElasticsearchClient, index: string) {
+  const { body } = await esClient.search<any>({
+    index,
+    size: 10000,
+    body: {
+      query: {
+        bool: {
+          should: [
+            {
+              term: { type: 'foo' },
+            },
+          ],
+        },
+      },
+    },
+  });
+
+  return body.hits.hits
+    .map((h) => ({
+      ...h._source,
+      id: h._id,
+    }))
+    .sort((a, b) => extractSortNumberFromId(a.id) - extractSortNumberFromId(b.id));
+}
+
+interface CreateRootConfig {
+  logFileName: string;
+}
+
+function createRoot({ logFileName }: CreateRootConfig) {
+  return kbnTestServer.createRoot(
+    {
+      elasticsearch: {
+        hosts: [esTestConfig.getUrl()],
+        username: kibanaServerTestUser.username,
+        password: kibanaServerTestUser.password,
+      },
+      migrations: {
+        skip: false,
+        enableV2: true,
+        batchSize: 100, // fixture contains 5000 docs
+      },
+      logging: {
+        appenders: {
+          file: {
+            type: 'file',
+            fileName: logFileName,
+            layout: {
+              type: 'pattern',
+            },
+          },
+        },
+        loggers: [
+          {
+            name: 'root',
+            appenders: ['file'],
+          },
+          {
+            name: 'savedobjects-service',
+            appenders: ['file'],
+            level: 'debug',
+          },
+        ],
+      },
+    },
+    {
+      oss: true,
+    }
+  );
+}
+
+describe('migration v2', () => {
+  let esServer: kbnTestServer.TestElasticsearchUtils;
+  let rootA: Root;
+  let rootB: Root;
+  let rootC: Root;
+
+  const migratedIndex = `.kibana_${pkg.version}_001`;
+  const fooType: SavedObjectsType = {
+    name: 'foo',
+    hidden: false,
+    mappings: { properties: { status: { type: 'text' } } },
+    namespaceType: 'agnostic',
+    migrations: {
+      '7.14.0': (doc) => {
+        if (doc.attributes?.status) {
+          doc.attributes.status = doc.attributes.status.replace('not_migrated', 'migrated');
+        }
+        return doc;
+      },
+    },
+  };
+
+  beforeAll(async () => {
+    await removeLogFiles();
+  });
+
+  afterAll(async () => {
+    await new Promise((resolve) => setTimeout(resolve, 10000));
+  });
+
+  beforeEach(async () => {
+    rootA = createRoot({
+      logFileName: Path.join(__dirname, `${LOG_FILE_PREFIX}_A.log`),
+    });
+    rootB = createRoot({
+      logFileName: Path.join(__dirname, `${LOG_FILE_PREFIX}_B.log`),
+    });
+    rootC = createRoot({
+      logFileName: Path.join(__dirname, `${LOG_FILE_PREFIX}_C.log`),
+    });
+
+    const { startES } = kbnTestServer.createTestServers({
+      adjustTimeout: (t: number) => jest.setTimeout(t),
+      settings: {
+        es: {
+          license: 'trial',
+          // original SO:
+          // [
+          //   { id: 'foo:1', type: 'foo', foo: { status: 'not_migrated_1' } },
+          //   { id: 'foo:2', type: 'foo', foo: { status: 'not_migrated_2' } },
+          //   { id: 'foo:3', type: 'foo', foo: { status: 'not_migrated_3' } },
+          // ];
+          dataArchive: Path.join(__dirname, 'archives', '7.13.0_5k_so.zip'),
+        },
+      },
+    });
+    esServer = await startES();
+  });
+
+  afterEach(async () => {
+    await Promise.all([rootA.shutdown(), rootB.shutdown(), rootC.shutdown()]);
+
+    if (esServer) {
+      await esServer.stop();
+    }
+  });
+
+  it('migrates saved objects normally when multiple Kibana instances are running', async () => {
+    const setupContracts = await Promise.all([rootA.setup(), rootB.setup(), rootC.setup()]);
+
+    setupContracts.forEach((setup) => setup.savedObjects.registerType(fooType));
+
+    const startContracts = await Promise.all([rootA.start(), rootB.start(), rootC.start()]);
+
+    const esClient = startContracts[0].elasticsearch.client.asInternalUser;
+    const migratedDocs = await fetchDocs(esClient, migratedIndex);
+
+    expect(migratedDocs.length).toBe(5000);
+    migratedDocs.forEach((doc, i) => {
+      expect(doc.id).toBe(`foo:${i}`);
+      expect(doc.foo.status).toBe(`migrated_${i}`);
+      expect(doc.migrationVersion.foo).toBe('7.14.0');
+    });
+  });
+
+  it('migrates saved objects normally when multiple Kibana instances are running with different plugins', async () => {
+    const setupContracts = await Promise.all([rootA.setup(), rootB.setup(), rootC.setup()]);
+
+    // Only register migration from one instance to simulate scenario
+    // where different plugins are enabled on different Kibana instances.
+    setupContracts[0].savedObjects.registerType(fooType);
+
+    // TODO: Fails if rootA.start promise isn't resolved first.
+    // Make sure this isn't an issue with the algorithm.
+    await rootA.start();
+    const startContracts = await Promise.all([rootB.start(), rootC.start()]);
+
+    const esClient = startContracts[0].elasticsearch.client.asInternalUser;
+    const migratedDocs = await fetchDocs(esClient, migratedIndex);
+
+    expect(migratedDocs.length).toBe(5000);
+    migratedDocs.forEach((doc, i) => {
+      expect(doc.id).toBe(`foo:${i}`);
+      expect(doc.foo.status).toBe(`migrated_${i}`);
+      expect(doc.migrationVersion.foo).toBe('7.14.0');
+    });
+  });
+});