[Reporting] roll over the reporting data stream when the template has changed (#234119)

resolves: #231200

This PR adds code run at startup to check the version in the reporting
index template against the `_meta.template_version` value stamped into
each backing index's mappings via
elastic/elasticsearch#133846

If it determines there are existing reporting indices that have not been
created with the current index template version, it will roll over the
reporting data stream, to ensure future indexing will use the latest
mappings. This is done with the "lazy" option, which will perform the
roll over on the next write to the datastream. This will prevent
situations where multiple Kibanas restarting at the same time would roll
over multiple times.

---------

Co-authored-by: Elastic Machine <[email protected]>

Author: pmuellr

.github/CODEOWNERS

@@ -1074,6 +1074,7 @@ x-pack/platform/test/plugin_api_integration/plugins/sample_task_plugin @elastic/
 x-pack/platform/test/plugin_api_perf/plugins/task_manager_performance @elastic/response-ops
 x-pack/platform/test/plugin_functional/plugins/global_search_test @elastic/kibana-core
 x-pack/platform/test/reporting_api_integration/plugins/reporting_fixture @elastic/response-ops
+x-pack/platform/test/reporting_api_integration/plugins/reporting_test_routes @elastic/response-ops
 x-pack/platform/test/saved_object_api_integration/common/plugins/saved_object_test_plugin @elastic/kibana-security
 x-pack/platform/test/security_api_integration/packages/helpers @elastic/kibana-security
 x-pack/platform/test/security_api_integration/plugins/audit_log @elastic/kibana-security

package.json

@@ -835,6 +835,7 @@
     "@kbn/reporting-plugin": "link:x-pack/platform/plugins/private/reporting",
     "@kbn/reporting-public": "link:src/platform/packages/private/kbn-reporting/public",
     "@kbn/reporting-server": "link:src/platform/packages/private/kbn-reporting/server",
+    "@kbn/reporting-test-routes": "link:x-pack/platform/test/reporting_api_integration/plugins/reporting_test_routes",
     "@kbn/resizable-layout": "link:src/platform/packages/shared/kbn-resizable-layout",
     "@kbn/resizable-layout-examples-plugin": "link:examples/resizable_layout_examples",
     "@kbn/resolver-test-plugin": "link:x-pack/solutions/security/test/plugin_functional/plugins/resolver_test",

src/platform/packages/private/kbn-reporting/server/constants.ts

@@ -23,6 +23,11 @@ export const REPORTING_LEGACY_INDICES = '.reporting-*';
 export const REPORTING_DATA_STREAM_WILDCARD_WITH_LEGACY = '.reporting-*,.kibana-reporting*';
 // Name of component template which Kibana overrides for lifecycle settings
 export const REPORTING_DATA_STREAM_COMPONENT_TEMPLATE = 'kibana-reporting@custom';
+// Name of index template
+export const REPORTING_DATA_STREAM_INDEX_TEMPLATE = '.kibana-reporting';
+// Name of mapping meta field which contains the version of the index template
+// see: https://github.com/elastic/elasticsearch/pull/133846
+export const REPORTING_INDEX_TEMPLATE_MAPPING_META_FIELD = 'template_version';
 
 /*
  * Telemetry

tsconfig.base.json

@@ -1672,6 +1672,8 @@
       "@kbn/reporting-public/*": ["src/platform/packages/private/kbn-reporting/public/*"],
       "@kbn/reporting-server": ["src/platform/packages/private/kbn-reporting/server"],
       "@kbn/reporting-server/*": ["src/platform/packages/private/kbn-reporting/server/*"],
+      "@kbn/reporting-test-routes": ["x-pack/platform/test/reporting_api_integration/plugins/reporting_test_routes"],
+      "@kbn/reporting-test-routes/*": ["x-pack/platform/test/reporting_api_integration/plugins/reporting_test_routes/*"],
       "@kbn/resizable-layout": ["src/platform/packages/shared/kbn-resizable-layout"],
       "@kbn/resizable-layout/*": ["src/platform/packages/shared/kbn-resizable-layout/*"],
       "@kbn/resizable-layout-examples-plugin": ["examples/resizable_layout_examples"],

x-pack/platform/plugins/private/reporting/server/lib/store/rollover.test.ts

@@ -0,0 +1,174 @@
+/*
+ * 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; you may not use this file except in compliance with the Elastic License
+ * 2.0.
+ */
+
+import { elasticsearchServiceMock } from '@kbn/core/server/mocks';
+import { loggingSystemMock } from '@kbn/core/server/mocks';
+import {
+  REPORTING_DATA_STREAM_ALIAS,
+  REPORTING_DATA_STREAM_INDEX_TEMPLATE,
+  REPORTING_INDEX_TEMPLATE_MAPPING_META_FIELD,
+} from '@kbn/reporting-server';
+import type {
+  IndicesGetIndexTemplateIndexTemplateItem,
+  IndicesGetMappingResponse,
+} from '@elastic/elasticsearch/lib/api/types';
+
+import { rollDataStreamIfRequired } from './rollover';
+
+describe('rollDataStreamIfRequired', () => {
+  const mockLogger = loggingSystemMock.createLogger();
+  let mockEsClient: ReturnType<typeof elasticsearchServiceMock.createElasticsearchClient>;
+
+  beforeEach(async () => {
+    mockEsClient = elasticsearchServiceMock.createElasticsearchClient();
+  });
+
+  const msgPrefix = `Data stream ${REPORTING_DATA_STREAM_ALIAS}`;
+  const skipMessage = 'does not need to be rolled over';
+  const rollMessage = 'rolling over the data stream';
+
+  beforeEach(async () => {
+    jest.clearAllMocks();
+  });
+
+  it('does nothing if there is no data stream', async () => {
+    mockEsClient.indices.exists.mockResponse(false);
+    await rollDataStreamIfRequired(mockLogger, mockEsClient);
+
+    expect(mockEsClient.indices.exists).toHaveBeenCalledWith({
+      index: REPORTING_DATA_STREAM_ALIAS,
+      expand_wildcards: 'all',
+    });
+    expect(mockLogger.debug).toHaveBeenCalledWith(`${msgPrefix} does not exist so ${skipMessage}`);
+    expect(mockEsClient.indices.getIndexTemplate).not.toHaveBeenCalled();
+    expect(mockEsClient.indices.getMapping).not.toHaveBeenCalled();
+    expect(mockEsClient.indices.rollover).not.toHaveBeenCalled();
+  });
+
+  it('throws an error if no index template is returned', async () => {
+    mockEsClient.indices.exists.mockResponse(true);
+    mockEsClient.indices.getIndexTemplate.mockResponse({ index_templates: [] });
+    const err = `${msgPrefix} index template ${REPORTING_DATA_STREAM_INDEX_TEMPLATE} not found`;
+    await expect(rollDataStreamIfRequired(mockLogger, mockEsClient)).rejects.toThrow(err);
+
+    expect(mockEsClient.indices.getIndexTemplate).toHaveBeenCalledWith({
+      name: REPORTING_DATA_STREAM_INDEX_TEMPLATE,
+    });
+    expect(mockEsClient.indices.getMapping).not.toHaveBeenCalled();
+    expect(mockEsClient.indices.rollover).not.toHaveBeenCalled();
+  });
+
+  it('throws an error if there is no index template with a version', async () => {
+    mockEsClient.indices.exists.mockResponse(true);
+    const templateWithoutVersion = getBasicIndexTemplate();
+    delete templateWithoutVersion.index_template.version;
+    mockEsClient.indices.getIndexTemplate.mockResponse({
+      index_templates: [templateWithoutVersion],
+    });
+
+    const err = `${msgPrefix} index template ${REPORTING_DATA_STREAM_INDEX_TEMPLATE} does not have a version field`;
+    await expect(rollDataStreamIfRequired(mockLogger, mockEsClient)).rejects.toThrow(err);
+
+    expect(mockEsClient.indices.getMapping).not.toHaveBeenCalled();
+    expect(mockEsClient.indices.rollover).not.toHaveBeenCalled();
+  });
+
+  it('does nothing if there are no mappings on the backing indices', async () => {
+    mockEsClient.indices.exists.mockResponse(true);
+    mockEsClient.indices.getIndexTemplate.mockResponse({
+      index_templates: [getBasicIndexTemplate()],
+    });
+    mockEsClient.indices.getMapping.mockResponse({});
+    await rollDataStreamIfRequired(mockLogger, mockEsClient);
+
+    const msg = `${msgPrefix} has no backing indices so ${skipMessage}`;
+    expect(mockLogger.debug).toHaveBeenCalledWith(msg);
+    expect(mockEsClient.indices.rollover).not.toHaveBeenCalled();
+  });
+
+  it('rolls over the data stream if there are no versions in the backing index mappings', async () => {
+    mockEsClient.indices.exists.mockResponse(true);
+    mockEsClient.indices.getIndexTemplate.mockResponse({
+      index_templates: [getBasicIndexTemplate()],
+    });
+    const mappings: IndicesGetMappingResponse = {
+      indexName: {
+        mappings: { _meta: {} },
+      },
+    };
+    mockEsClient.indices.getMapping.mockResponse(mappings);
+    await rollDataStreamIfRequired(mockLogger, mockEsClient);
+
+    const msg = `${msgPrefix} has no mapping versions so ${rollMessage}`;
+    expect(mockLogger.info).toHaveBeenCalledWith(msg);
+    expect(mockEsClient.indices.rollover).toHaveBeenCalled();
+  });
+
+  it('rolls over the data stream if the index template version is newer than the backing index mappings versions', async () => {
+    mockEsClient.indices.exists.mockResponse(true);
+    mockEsClient.indices.getIndexTemplate.mockResponse({
+      index_templates: [getBasicIndexTemplate()],
+    });
+    const mappings: IndicesGetMappingResponse = {
+      indexName: {
+        mappings: { _meta: { [REPORTING_INDEX_TEMPLATE_MAPPING_META_FIELD]: 41 } },
+      },
+    };
+    mockEsClient.indices.getMapping.mockResponse(mappings);
+    await rollDataStreamIfRequired(mockLogger, mockEsClient);
+
+    const msg = `${msgPrefix} has older mappings than the template so ${rollMessage}`;
+    expect(mockLogger.info).toHaveBeenCalledWith(msg);
+    expect(mockEsClient.indices.rollover).toHaveBeenCalled();
+  });
+
+  it('throws an error if the index template version is older than the backing index mappings versions', async () => {
+    mockEsClient.indices.exists.mockResponse(true);
+    mockEsClient.indices.getIndexTemplate.mockResponse({
+      index_templates: [getBasicIndexTemplate()],
+    });
+    const mappings: IndicesGetMappingResponse = {
+      indexName: {
+        mappings: { _meta: { [REPORTING_INDEX_TEMPLATE_MAPPING_META_FIELD]: 43 } },
+      },
+    };
+    mockEsClient.indices.getMapping.mockResponse(mappings);
+    const err = `${msgPrefix} has newer mappings than the template`;
+    await expect(rollDataStreamIfRequired(mockLogger, mockEsClient)).rejects.toThrow(err);
+
+    expect(mockEsClient.indices.rollover).not.toHaveBeenCalled();
+  });
+
+  it('does nothing if the index template version is not newer than the backing index mapping versions', async () => {
+    mockEsClient.indices.exists.mockResponse(true);
+    mockEsClient.indices.getIndexTemplate.mockResponse({
+      index_templates: [getBasicIndexTemplate()],
+    });
+    const mappings: IndicesGetMappingResponse = {
+      indexName: {
+        mappings: { _meta: { [REPORTING_INDEX_TEMPLATE_MAPPING_META_FIELD]: 42 } },
+      },
+    };
+    mockEsClient.indices.getMapping.mockResponse(mappings);
+    await rollDataStreamIfRequired(mockLogger, mockEsClient);
+
+    const msg = `${msgPrefix} has latest mappings applied so ${skipMessage}`;
+    expect(mockLogger.debug).toHaveBeenCalledWith(msg);
+    expect(mockEsClient.indices.rollover).not.toHaveBeenCalled();
+  });
+});
+
+function getBasicIndexTemplate(): IndicesGetIndexTemplateIndexTemplateItem {
+  return {
+    name: REPORTING_DATA_STREAM_INDEX_TEMPLATE,
+    index_template: {
+      index_patterns: ['ignored'],
+      composed_of: ['ignored'],
+      version: 42,
+    },
+  };
+}

x-pack/platform/plugins/private/reporting/server/lib/store/rollover.ts

@@ -0,0 +1,105 @@
+/*
+ * 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; you may not use this file except in compliance with the Elastic License
+ * 2.0.
+ */
+
+import {
+  REPORTING_DATA_STREAM_ALIAS,
+  REPORTING_DATA_STREAM_INDEX_TEMPLATE,
+  REPORTING_INDEX_TEMPLATE_MAPPING_META_FIELD,
+} from '@kbn/reporting-server';
+import type { ElasticsearchClient, Logger } from '@kbn/core/server';
+
+export async function rollDataStreamIfRequired(
+  logger: Logger,
+  esClient: ElasticsearchClient
+): Promise<boolean> {
+  const msgPrefix = `Data stream ${REPORTING_DATA_STREAM_ALIAS}`;
+  const skipMessage = 'does not need to be rolled over';
+  const rollMessage = 'rolling over the data stream';
+  // easy way to change debug log level when debugging
+  const debug = (msg: string) => logger.debug(msg);
+
+  const exists = await esClient.indices.exists({
+    index: REPORTING_DATA_STREAM_ALIAS,
+    expand_wildcards: 'all',
+  });
+
+  if (!exists) {
+    debug(`${msgPrefix} does not exist so ${skipMessage}`);
+    return false;
+  }
+
+  const gotTemplate = await esClient.indices.getIndexTemplate({
+    name: REPORTING_DATA_STREAM_INDEX_TEMPLATE,
+  });
+  if (gotTemplate.index_templates.length === 0) {
+    throw new Error(
+      `${msgPrefix} index template ${REPORTING_DATA_STREAM_INDEX_TEMPLATE} not found`
+    );
+  }
+
+  const templateVersions: number[] = [];
+  for (const template of gotTemplate.index_templates) {
+    const templateVersion = template.index_template.version;
+    if (templateVersion) templateVersions.push(templateVersion);
+  }
+
+  if (templateVersions.length === 0) {
+    throw new Error(
+      `${msgPrefix} index template ${REPORTING_DATA_STREAM_INDEX_TEMPLATE} does not have a version field`
+    );
+  }
+
+  // assume the highest version is the one in use
+  const templateVersion = Math.max(...templateVersions);
+  debug(`${msgPrefix} template version: ${templateVersion}`);
+
+  const mappings = await esClient.indices.getMapping({
+    index: REPORTING_DATA_STREAM_ALIAS,
+    allow_no_indices: true,
+    expand_wildcards: 'all',
+  });
+
+  const mappingsArray = Object.values(mappings);
+  if (mappingsArray.length === 0) {
+    debug(`${msgPrefix} has no backing indices so ${skipMessage}`);
+    return false;
+  }
+
+  // get the value of _meta.template_version from each index's mappings
+  const mappingsVersions = mappingsArray
+    .map((m) => m.mappings._meta?.[REPORTING_INDEX_TEMPLATE_MAPPING_META_FIELD])
+    .filter((a: any): a is number => typeof a === 'number');
+
+  const mappingsVersion = mappingsVersions.length === 0 ? undefined : Math.max(...mappingsVersions);
+  debug(`${msgPrefix} mappings version: ${mappingsVersion ?? '<none>'}`);
+
+  if (mappingsVersion === undefined) {
+    // no mapping version found on any indices
+    logger.info(`${msgPrefix} has no mapping versions so ${rollMessage}`);
+  } else if (mappingsVersion < templateVersion) {
+    // all mappings are old
+    logger.info(`${msgPrefix} has older mappings than the template so ${rollMessage}`);
+  } else if (mappingsVersion > templateVersion) {
+    // newer mappings than the template shouldn't happen
+    throw new Error(`${msgPrefix} has newer mappings than the template`);
+  } else {
+    // latest mappings already applied
+    debug(`${msgPrefix} has latest mappings applied so ${skipMessage}`);
+    return false;
+  }
+
+  // Roll over the data stream to pick up the new mappings.
+  // The `lazy` option will cause the rollover to run on the next write.
+  // This limits potential race conditions of multiple Kibana's rolling over at once.
+  await esClient.indices.rollover({
+    alias: REPORTING_DATA_STREAM_ALIAS,
+    lazy: true,
+  });
+
+  logger.info(`${msgPrefix} rolled over to pick up index template version ${templateVersion}`);
+  return true;
+}

x-pack/platform/plugins/private/reporting/server/lib/store/store.ts

@@ -25,6 +25,7 @@ import type { ReportingCore } from '../..';
 import type { ReportTaskParams } from '../tasks';
 import { IlmPolicyManager } from './ilm_policy_manager';
 import { MIGRATION_VERSION } from './report';
+import { rollDataStreamIfRequired } from './rollover';
 
 type UpdateResponse<T> = estypes.UpdateResponse<T>;
 type IndexResponse = estypes.IndexResponse;
@@ -170,10 +171,20 @@ export class ReportingStore {
         await this.createIlmPolicy();
       }
     } catch (e) {
-      this.logger.error('Error in start phase');
-      this.logger.error(e);
+      this.logger.error(`Error creating ILM policy: ${e.message}`, {
+        error: { stack_trace: e.stack },
+      });
       throw e;
     }
+
+    try {
+      await rollDataStreamIfRequired(this.logger, await this.getClient());
+    } catch (e) {
+      this.logger.error(`Error rolling over data stream: ${e.message}`, {
+        error: { stack_trace: e.stack },
+      });
+      // not rethrowing, as this is not a fatal error
+    }
   }
 
   public async addReport(report: Report): Promise<SavedReport> {

x-pack/platform/test/reporting_api_integration/plugins/reporting_test_routes/kibana.jsonc

@@ -0,0 +1,16 @@
+{
+  "type": "plugin",
+  "id": "@kbn/reporting-test-routes",
+  "owner": "@elastic/response-ops",
+  "visibility": "private",
+  "plugin": {
+    "id": "reportingTestRoutes",
+    "server": true,
+    "browser": false,
+    "requiredPlugins": [
+      "reporting",
+    ],
+    "optionalPlugins": [
+    ]
+  }
+}

x-pack/platform/test/reporting_api_integration/plugins/reporting_test_routes/package.json

@@ -0,0 +1,14 @@
+{
+  "name": "@kbn/reporting-test-routes",
+  "version": "1.0.0",
+  "kibana": {
+    "version": "kibana",
+    "templateVersion": "1.0.0"
+  },
+  "main": "target/test/reporting_api_integration/plugins/reporting_api_integration",
+  "scripts": {
+    "kbn": "node ../../../../../../scripts/kbn.js",
+    "build": "rm -rf './target' && ../../../../../../node_modules/.bin/tsc"
+  },
+  "license": "Elastic License 2.0"
+}

x-pack/platform/test/reporting_api_integration/plugins/reporting_test_routes/server/index.ts

@@ -0,0 +1,11 @@
+/*
+ * 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; you may not use this file except in compliance with the Elastic License
+ * 2.0.
+ */
+
+import type { PluginInitializerContext } from '@kbn/core/server';
+import { TestPlugin } from './plugin';
+
+export const plugin = async (initContext: PluginInitializerContext) => new TestPlugin(initContext);