[ftr] implement support for accessing ES through CCS

packages/kbn-test/BUILD.bazel

@@ -53,6 +53,7 @@ RUNTIME_DEPS = [
     "@npm//execa",
     "@npm//exit-hook",
     "@npm//form-data",
+    "@npm//get-port",
     "@npm//getopts",
     "@npm//globby",
     "@npm//he",
@@ -90,6 +91,7 @@ TYPES_DEPS = [
   "@npm//del",
   "@npm//exit-hook",
   "@npm//form-data",
+  "@npm//get-port",
   "@npm//getopts",
   "@npm//jest",
   "@npm//jest-cli",

packages/kbn-test/src/es/es_client_for_testing.ts

@@ -27,6 +27,22 @@ export interface EsClientForTestingOptions extends Omit<ClientOptions, 'node' |
   isCloud?: boolean;
 }
 
+export function createRemoteEsClientForFtrConfig(
+  config: Config,
+  overrides?: Omit<EsClientForTestingOptions, 'esUrl'>
+) {
+  const ccsConfig = config.get('esTestCluster.ccs');
+  if (!ccsConfig) {
+    throw new Error('FTR config is missing esTestCluster.ccs');
+  }
+
+  return createEsClientForTesting({
+    esUrl: ccsConfig.remoteClusterUrl,
+    requestTimeout: config.get('timeouts.esRequestTimeout'),
+    ...overrides,
+  });
+}
+
 export function createEsClientForFtrConfig(
   config: Config,
   overrides?: Omit<EsClientForTestingOptions, 'esUrl'>

packages/kbn-test/src/es/index.ts

@@ -9,5 +9,9 @@
 export { createTestEsCluster } from './test_es_cluster';
 export type { CreateTestEsClusterOptions, EsTestCluster, ICluster } from './test_es_cluster';
 export { esTestConfig } from './es_test_config';
-export { createEsClientForTesting, createEsClientForFtrConfig } from './es_client_for_testing';
+export {
+  createEsClientForTesting,
+  createEsClientForFtrConfig,
+  createRemoteEsClientForFtrConfig,
+} from './es_client_for_testing';
 export type { EsClientForTestingOptions } from './es_client_for_testing';

packages/kbn-test/src/es/test_es_cluster.ts

@@ -136,7 +136,15 @@ export interface CreateTestEsClusterOptions {
    * }
    */
   port?: number;
+  /**
+   * Should this ES cluster use SSL?
+   */
   ssl?: boolean;
+  /**
+   * Explicit transport port for a single node to run on, or a string port range to use eg. '9300-9400'
+   * defaults to the transport port from `packages/kbn-test/src/es/es_test_config.ts`
+   */
+  transportPort?: number | string;
 }
 
 export function createTestEsCluster<
@@ -155,13 +163,14 @@ export function createTestEsCluster<
     esJavaOpts,
     clusterName: customClusterName = 'es-test-cluster',
     ssl,
+    transportPort,
   } = options;
 
   const clusterName = `${CI_PARALLEL_PROCESS_PREFIX}${customClusterName}`;
 
   const defaultEsArgs = [
     `cluster.name=${clusterName}`,
-    `transport.port=${esTestConfig.getTransportPort()}`,
+    `transport.port=${transportPort ?? esTestConfig.getTransportPort()}`,
     // For multi-node clusters, we make all nodes master-eligible by default.
     ...(nodes.length > 1
       ? ['discovery.type=zen', `cluster.initial_master_nodes=${nodes.map((n) => n.name).join(',')}`]

packages/kbn-test/src/functional_test_runner/lib/config/schema.ts

@@ -192,12 +192,17 @@ export const schema = Joi.object()
 
     esTestCluster: Joi.object()
       .keys({
-        license: Joi.string().default('basic'),
+        license: Joi.valid('basic', 'trial', 'gold').default('basic'),
         from: Joi.string().default('snapshot'),
-        serverArgs: Joi.array(),
+        serverArgs: Joi.array().items(Joi.string()),
         esJavaOpts: Joi.string(),
         dataArchive: Joi.string(),
         ssl: Joi.boolean().default(false),
+        ccs: Joi.object().keys({
+          remoteClusterUrl: Joi.string().uri({
+            scheme: /https?/,
+          }),
+        }),
       })
       .default(),
 
@@ -290,6 +295,7 @@ export const schema = Joi.object()
     security: Joi.object()
       .keys({
         roles: Joi.object().default(),
+        remoteEsRoles: Joi.object(),
         defaultRoles: Joi.array()
           .items(Joi.string())
           .when('$primary', {

packages/kbn-test/src/functional_tests/lib/run_elasticsearch.ts

@@ -8,39 +8,110 @@
 
 import { resolve } from 'path';
 import type { ToolingLog } from '@kbn/dev-utils';
+import getPort from 'get-port';
 import { KIBANA_ROOT } from './paths';
 import type { Config } from '../../functional_test_runner/';
 import { createTestEsCluster } from '../../es';
 
 interface RunElasticsearchOptions {
   log: ToolingLog;
   esFrom?: string;
+  config: Config;
+}
+
+interface CcsConfig {
+  remoteClusterUrl: string;
 }
-export async function runElasticsearch({
+
+type EsConfig = ReturnType<typeof getEsConfig>;
+
+function getEsConfig({
   config,
-  options,
-}: {
-  config: Config;
-  options: RunElasticsearchOptions;
-}) {
-  const { log, esFrom } = options;
-  const ssl = config.get('esTestCluster.ssl');
-  const license = config.get('esTestCluster.license');
-  const esArgs = config.get('esTestCluster.serverArgs');
-  const esJavaOpts = config.get('esTestCluster.esJavaOpts');
+  esFrom = config.get('esTestCluster.from'),
+}: RunElasticsearchOptions) {
+  const ssl = !!config.get('esTestCluster.ssl');
+  const license: 'basic' | 'trial' | 'gold' = config.get('esTestCluster.license');
+  const esArgs: string[] = config.get('esTestCluster.serverArgs') ?? [];
+  const esJavaOpts: string | undefined = config.get('esTestCluster.esJavaOpts');
   const isSecurityEnabled = esArgs.includes('xpack.security.enabled=true');
 
-  const cluster = createTestEsCluster({
-    port: config.get('servers.elasticsearch.port'),
-    password: isSecurityEnabled ? 'changeme' : config.get('servers.elasticsearch.password'),
+  const port: number | undefined = config.get('servers.elasticsearch.port');
+  const ccsConfig: CcsConfig | undefined = config.get('esTestCluster.ccs');
+
+  const password: string | undefined = isSecurityEnabled
+    ? 'changeme'
+    : config.get('servers.elasticsearch.password');
+
+  const dataArchive: string | undefined = config.get('esTestCluster.dataArchive');
+
+  return {
+    ssl,
     license,
-    log,
-    basePath: resolve(KIBANA_ROOT, '.es'),
-    esFrom: esFrom || config.get('esTestCluster.from'),
-    dataArchive: config.get('esTestCluster.dataArchive'),
     esArgs,
     esJavaOpts,
-    ssl,
+    isSecurityEnabled,
+    esFrom,
+    port,
+    password,
+    dataArchive,
+    ccsConfig,
+  };
+}
+
+export async function runElasticsearch(
+  options: RunElasticsearchOptions
+): Promise<() => Promise<void>> {
+  const { log } = options;
+  const config = getEsConfig(options);
+
+  if (!config.ccsConfig) {
+    const node = await startEsNode(log, 'ftr', config);
+    return async () => {
+      await node.cleanup();
+    };
+  }
+
+  const remotePort = await getPort();
+  const remoteNode = await startEsNode(log, 'ftr-remote', {
+    ...config,
+    port: parseInt(new URL(config.ccsConfig.remoteClusterUrl).port, 10),
+    transportPort: remotePort,
+  });
+
+  const localNode = await startEsNode(log, 'ftr-local', {
+    ...config,
+    esArgs: [...config.esArgs, `cluster.remote.ftr-remote.seeds=localhost:${remotePort}`],
+  });
+
+  return async () => {
+    await localNode.cleanup();
+    await remoteNode.cleanup();
+  };
+}
+
+async function startEsNode(
+  log: ToolingLog,
+  name: string,
+  config: EsConfig & { transportPort?: number }
+) {
+  const cluster = createTestEsCluster({
+    clusterName: `cluster-${name}`,
+    esArgs: config.esArgs,
+    esFrom: config.esFrom,
+    esJavaOpts: config.esJavaOpts,
+    license: config.license,
+    password: config.password,
+    port: config.port,
+    ssl: config.ssl,
+    log,
+    basePath: resolve(KIBANA_ROOT, '.es'),
+    nodes: [
+      {
+        name,
+        dataArchive: config.dataArchive,
+      },
+    ],
+    transportPort: config.transportPort,
   });
 
   await cluster.start();

packages/kbn-test/src/functional_tests/tasks.ts

@@ -108,10 +108,10 @@ export async function runTests(options: RunTestsParams) {
       await withProcRunner(log, async (procs) => {
         const config = await readConfigFile(log, options.esVersion, configPath);
 
-        let es;
+        let shutdownEs;
         try {
           if (process.env.TEST_ES_DISABLE_STARTUP !== 'true') {
-            es = await runElasticsearch({ config, options: { ...options, log } });
+            shutdownEs = await runElasticsearch({ ...options, log, config });
           }
           await runKibanaServer({ procs, config, options });
           await runFtr({ configPath, options: { ...options, log } });
@@ -125,8 +125,8 @@ export async function runTests(options: RunTestsParams) {
 
             await procs.stop('kibana');
           } finally {
-            if (es) {
-              await es.cleanup();
+            if (shutdownEs) {
+              await shutdownEs();
             }
           }
         }
@@ -166,7 +166,7 @@ export async function startServers({ ...options }: StartServerOptions) {
   await withProcRunner(log, async (procs) => {
     const config = await readConfigFile(log, options.esVersion, options.config);
 
-    const es = await runElasticsearch({ config, options: opts });
+    const shutdownEs = await runElasticsearch({ ...opts, config });
     await runKibanaServer({
       procs,
       config,
@@ -190,7 +190,7 @@ export async function startServers({ ...options }: StartServerOptions) {
     log.success(makeSuccessMessage(options));
 
     await procs.waitForAllToStop();
-    await es.cleanup();
+    await shutdownEs();
   });
 }
 

packages/kbn-test/src/index.ts

@@ -36,6 +36,7 @@ export {
   createTestEsCluster,
   createEsClientForTesting,
   createEsClientForFtrConfig,
+  createRemoteEsClientForFtrConfig,
 } from './es';
 
 export {

scripts/functional_tests.js

@@ -9,7 +9,7 @@
 require('../src/setup_node_env');
 require('@kbn/test').runTestsCli([
   require.resolve('../test/functional/config.js'),
-  require.resolve('../test/functional_ccs/config.js'),
+  require.resolve('../test/functional_ccs/config.ts'),
   require.resolve('../test/plugin_functional/config.ts'),
   require.resolve('../test/ui_capabilities/newsfeed_err/config.ts'),
   require.resolve('../test/new_visualize_flow/config.ts'),

test/common/services/es_archiver.ts

@@ -8,11 +8,12 @@
 
 import { EsArchiver } from '@kbn/es-archiver';
 import { FtrProviderContext } from '../ftr_provider_context';
-import * as KibanaServer from './kibana_server';
+import * as KibanaServer from '../../common/services/kibana_server';
 
 export function EsArchiverProvider({ getService }: FtrProviderContext): EsArchiver {
   const config = getService('config');
   const client = getService('es');
+
   const log = getService('log');
   const kibanaServer = getService('kibanaServer');
   const retry = getService('retry');

test/common/services/security/system_indices_user.ts

@@ -6,25 +6,18 @@
  * Side Public License, v 1.
  */
 
-import { systemIndicesSuperuser, createEsClientForFtrConfig } from '@kbn/test';
+import { Client } from '@elastic/elasticsearch';
+import { ToolingLog } from '@kbn/dev-utils';
+import {
+  systemIndicesSuperuser,
+  createEsClientForFtrConfig,
+  createRemoteEsClientForFtrConfig,
+} from '@kbn/test';
 import { FtrProviderContext } from '../../ftr_provider_context';
 
 const SYSTEM_INDICES_SUPERUSER_ROLE = 'system_indices_superuser';
 
-export async function createSystemIndicesUser(ctx: FtrProviderContext) {
-  const log = ctx.getService('log');
-  const config = ctx.getService('config');
-
-  const enabled = !config
-    .get('esTestCluster.serverArgs')
-    .some((arg: string) => arg === 'xpack.security.enabled=false');
-
-  if (!enabled) {
-    return;
-  }
-
-  const es = createEsClientForFtrConfig(config);
-
+async function ensureSystemIndicesUser(es: Client, log: ToolingLog) {
   // There are cases where the test config file doesn't have security disabled
   // but tests are still executed on ES without security. Checking this case
   // by trying to fetch the users list.
@@ -67,3 +60,24 @@ export async function createSystemIndicesUser(ctx: FtrProviderContext) {
 
   await es.close();
 }
+
+export async function createSystemIndicesUser(ctx: FtrProviderContext) {
+  const log = ctx.getService('log');
+  const config = ctx.getService('config');
+
+  const enabled = !config
+    .get('esTestCluster.serverArgs')
+    .some((arg: string) => arg === 'xpack.security.enabled=false');
+
+  if (!enabled) {
+    return;
+  }
+
+  const localEs = createEsClientForFtrConfig(config);
+  await ensureSystemIndicesUser(localEs, log);
+
+  if (config.get('esTestCluster.ccs')) {
+    const remoteEs = createRemoteEsClientForFtrConfig(config);
+    await ensureSystemIndicesUser(remoteEs, log);
+  }
+}