[Reporting] roll over the reporting data stream when the template has changed

.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);