[Response Ops][Event Log] Updating event log mappings if data stream …

…and index template already exist (elastic#193205)

Resolves elastic#192682

## Summary

As of 8.8, we started writing all event log documents to the
`.kibana-event-log-ds` index. Prior to this, we created a new index
template and data stream for every version (`.kibana-event-log-8.7` for
example) so any mapping updates that were added for the version were
created in the new index on upgrade.

With the static index name and serverless, we need a way to update
mappings in existing indices. This PR uses the same mechanism that we
use for the alerts index to update the index template mappings and the
mappings for the concrete backing indices of a datastream.

## To Verify

Run ES and Kibana in `main` to test the upgrade path for serverless 
a. Check out `main`, run ES: `yarn es snapshot --license trial --ssl -E
path.data=../test_el_upgrade` and Kibana `yarn start --ssl`
  b. Create a rule and let it run to populate the event log index
c. Switch to this PR branch. Make a mapping update to the event log
index:

```
--- a/x-pack/plugins/event_log/generated/mappings.json
+++ b/x-pack/plugins/event_log/generated/mappings.json
@@ -172,6 +172,9 @@
         },
         "rule": {
             "properties": {
+                "test": {
+                  "type": "keyword"
+                },
                 "author": {
                     "ignore_above": 1024,
                     "type": "keyword",
```
  d. Start ES and Kibana with the same commands as above
e. Verify that the `.kibana-event-log-ds` index is created and has the
updated mapping:
-
https://localhost:5601/app/management/data/index_management/templates/.kibana-event-log-template
-
https://localhost:5601/app/management/data/index_management/indices/index_details?indexName=.ds-.kibana-event-log-ds-2024.09.17-000001&filter=.kibana-&includeHiddenIndices=true&tab=mappings

I also verified the following:
1. Run ES and Kibana in 8.7 to test the upgrade path from 8.7 (when
event log indices were versioned) to now
2. Run ES and Kibana in 8.15 to test the upgrade path from the previous
release to now

However, I had to create an 8.x branch and cherry pick this commit
because `main` is now on 9.0 and we can't upgrade directly from older
8.x version to 9.0!

---------

Co-authored-by: Elastic Machine <elasticmachine@users.noreply.github.com>
Co-authored-by: kibanamachine <42973632+kibanamachine@users.noreply.github.com>

x-pack/plugins/event_log/jest.integration.config.js

@@ -0,0 +1,12 @@
+/*
+ * 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.
+ */
+
+module.exports = {
+ preset: '@kbn/test/jest_integration',
+ rootDir: '../../..',
+ roots: ['<rootDir>/x-pack/plugins/event_log'],
+};

x-pack/plugins/event_log/server/es/cluster_client_adapter.mock.ts

@@ -13,8 +13,10 @@ const createClusterClientMock = () => {
  indexDocuments: jest.fn(),
  doesIndexTemplateExist: jest.fn(),
  createIndexTemplate: jest.fn(),
+ updateIndexTemplate: jest.fn(),
  doesDataStreamExist: jest.fn(),
  createDataStream: jest.fn(),
+ updateConcreteIndices: jest.fn(),
  getExistingLegacyIndexTemplates: jest.fn(),
  setLegacyIndexTemplateToHidden: jest.fn(),
  getExistingIndices: jest.fn(),

x-pack/plugins/event_log/server/es/cluster_client_adapter.test.ts

@@ -215,6 +215,117 @@ describe('createIndexTemplate', () => {
  });
 });
 
+describe('updateIndexTemplate', () => {
+ test('should call cluster with given template', async () => {
+ clusterClient.indices.simulateTemplate.mockImplementationOnce(async () => ({
+ template: {
+ aliases: {
+ alias_name_1: {
+ is_hidden: true,
+ },
+ alias_name_2: {
+ is_hidden: true,
+ },
+ },
+ settings: {
+ hidden: true,
+ number_of_shards: 1,
+ auto_expand_replicas: '0-1',
+ },
+ mappings: { dynamic: false, properties: { '@timestamp': { type: 'date' } } },
+ },
+ }));
+
+ await clusterClientAdapter.updateIndexTemplate('foo', { args: true });
+
+ expect(clusterClient.indices.simulateTemplate).toHaveBeenCalledWith({
+ name: 'foo',
+ body: { args: true },
+ });
+ expect(clusterClient.indices.putIndexTemplate).toHaveBeenCalledWith({
+ name: 'foo',
+ body: { args: true },
+ });
+ });
+
+ test(`should throw error if simulate mappings response is empty`, async () => {
+ clusterClient.indices.simulateTemplate.mockImplementationOnce(async () => ({
+ template: {
+ aliases: {
+ alias_name_1: {
+ is_hidden: true,
+ },
+ alias_name_2: {
+ is_hidden: true,
+ },
+ },
+ settings: {
+ hidden: true,
+ number_of_shards: 1,
+ auto_expand_replicas: '0-1',
+ },
+ mappings: {},
+ },
+ }));
+
+ await expect(() =>
+ clusterClientAdapter.updateIndexTemplate('foo', { name: 'template', args: true })
+ ).rejects.toThrowErrorMatchingInlineSnapshot(
+ `"No mappings would be generated for template, possibly due to failed/misconfigured bootstrapping"`
+ );
+
+ expect(logger.error).toHaveBeenCalledWith(
+ `Error updating index template foo: No mappings would be generated for template, possibly due to failed/misconfigured bootstrapping`
+ );
+ });
+
+ test(`should throw error if simulateTemplate throws error`, async () => {
+ clusterClient.indices.simulateTemplate.mockImplementationOnce(() => {
+ throw new Error('failed to simulate');
+ });
+
+ await expect(() =>
+ clusterClientAdapter.updateIndexTemplate('foo', { name: 'template', args: true })
+ ).rejects.toThrowErrorMatchingInlineSnapshot(`"failed to simulate"`);
+
+ expect(logger.error).toHaveBeenCalledWith(
+ `Error updating index template foo: failed to simulate`
+ );
+ });
+
+ test(`should throw error if putIndexTemplate throws error`, async () => {
+ clusterClient.indices.simulateTemplate.mockImplementationOnce(async () => ({
+ template: {
+ aliases: {
+ alias_name_1: {
+ is_hidden: true,
+ },
+ alias_name_2: {
+ is_hidden: true,
+ },
+ },
+ settings: {
+ hidden: true,
+ number_of_shards: 1,
+ auto_expand_replicas: '0-1',
+ },
+ mappings: { dynamic: false, properties: { '@timestamp': { type: 'date' } } },
+ },
+ }));
+ clusterClient.indices.putIndexTemplate.mockImplementationOnce(() => {
+ throw new Error('failed to update index template');
+ });
+
+ await expect(() =>
+ clusterClientAdapter.updateIndexTemplate('foo', { name: 'template', args: true })
+ ).rejects.toThrowErrorMatchingInlineSnapshot(`"failed to update index template"`);
+
+ expect(logger.error).toHaveBeenCalledWith(
+ `Error updating index template foo: failed to update index template`
+ );
+ });
+});
+
 describe('getExistingLegacyIndexTemplates', () => {
  test('should call cluster with given index template pattern', async () => {
  await clusterClientAdapter.getExistingLegacyIndexTemplates('foo*');
@@ -497,7 +608,7 @@ describe('doesDataStreamExist', () => {
  });
 });
 
-describe('createIndex', () => {
+describe('createDataStream', () => {
  test('should call cluster with proper arguments', async () => {
  await clusterClientAdapter.createDataStream('foo');
  expect(clusterClient.indices.createDataStream).toHaveBeenCalledWith({
@@ -526,6 +637,95 @@ describe('createIndex', () => {
  });
 });
 
+describe('updateConcreteIndices', () => {
+ test('should call cluster with proper arguments', async () => {
+ clusterClient.indices.simulateIndexTemplate.mockImplementationOnce(async () => ({
+ template: {
+ aliases: { alias_name_1: { is_hidden: true } },
+ settings: {
+ hidden: true,
+ number_of_shards: 1,
+ auto_expand_replicas: '0-1',
+ },
+ mappings: { dynamic: false, properties: { '@timestamp': { type: 'date' } } },
+ },
+ }));
+
+ await clusterClientAdapter.updateConcreteIndices('foo');
+ expect(clusterClient.indices.simulateIndexTemplate).toHaveBeenCalledWith({
+ name: 'foo',
+ });
+ expect(clusterClient.indices.putMapping).toHaveBeenCalledWith({
+ index: 'foo',
+ body: { dynamic: false, properties: { '@timestamp': { type: 'date' } } },
+ });
+ });
+
+ test('should not update mapping if simulate response does not contain mappings', async () => {
+ // @ts-ignore
+ clusterClient.indices.simulateIndexTemplate.mockImplementationOnce(async () => ({
+ template: {
+ aliases: { alias_name_1: { is_hidden: true } },
+ settings: {
+ hidden: true,
+ number_of_shards: 1,
+ auto_expand_replicas: '0-1',
+ },
+ },
+ }));
+
+ await clusterClientAdapter.updateConcreteIndices('foo');
+ expect(clusterClient.indices.simulateIndexTemplate).toHaveBeenCalledWith({
+ name: 'foo',
+ });
+ expect(clusterClient.indices.putMapping).not.toHaveBeenCalled();
+ });
+
+ test('should throw error if simulateIndexTemplate throws error', async () => {
+ clusterClient.indices.simulateIndexTemplate.mockImplementationOnce(() => {
+ throw new Error('failed to simulate');
+ });
+
+ await expect(() =>
+ clusterClientAdapter.updateConcreteIndices('foo')
+ ).rejects.toThrowErrorMatchingInlineSnapshot(`"failed to simulate"`);
+
+ expect(clusterClient.indices.putMapping).not.toHaveBeenCalled();
+ expect(logger.error).toHaveBeenCalledWith(
+ `Error updating index mappings for foo: failed to simulate`
+ );
+ });
+
+ test('should throw error if putMapping throws error', async () => {
+ clusterClient.indices.simulateIndexTemplate.mockImplementationOnce(async () => ({
+ template: {
+ aliases: { alias_name_1: { is_hidden: true } },
+ settings: {
+ hidden: true,
+ number_of_shards: 1,
+ auto_expand_replicas: '0-1',
+ },
+ mappings: { dynamic: false, properties: { '@timestamp': { type: 'date' } } },
+ },
+ }));
+ clusterClient.indices.putMapping.mockImplementationOnce(() => {
+ throw new Error('failed to put mappings');
+ });
+
+ await expect(() =>
+ clusterClientAdapter.updateConcreteIndices('foo')
+ ).rejects.toThrowErrorMatchingInlineSnapshot(`"failed to put mappings"`);
+
+ expect(clusterClient.indices.putMapping).toHaveBeenCalledWith({
+ index: 'foo',
+ body: { dynamic: false, properties: { '@timestamp': { type: 'date' } } },
+ });
+ expect(logger.error).toHaveBeenCalledWith(
+ `Error updating index mappings for foo: failed to put mappings`
+ );
+ });
+});
+
 describe('queryEventsBySavedObject', () => {
  const DEFAULT_OPTIONS = queryOptionsSchema.validate({});
 

x-pack/plugins/event_log/server/es/cluster_client_adapter.ts

@@ -7,7 +7,7 @@
 
 import { Subject } from 'rxjs';
 import { bufferTime, filter as rxFilter, concatMap } from 'rxjs';
-import { reject, isUndefined, isNumber, pick } from 'lodash';
+import { reject, isUndefined, isNumber, pick, isEmpty, get } from 'lodash';
 import type { PublicMethodsOf } from '@kbn/utility-types';
 import { Logger, ElasticsearchClient } from '@kbn/core/server';
 import util from 'util';
@@ -213,6 +213,28 @@ export class ClusterClientAdapter<TDoc extends { body: AliasAny; index: string }
  }
  }
 
+ public async updateIndexTemplate(name: string, template: Record<string, unknown>): Promise<void> {
+ this.logger.info(`Updating index template ${name}`);
+
+ try {
+ const esClient = await this.elasticsearchClientPromise;
+
+ // Simulate the index template to proactively identify any issues with the mappings
+ const simulateResponse = await esClient.indices.simulateTemplate({ name, body: template });
+ const mappings: estypes.MappingTypeMapping = simulateResponse.template.mappings;
+
+ if (isEmpty(mappings)) {
+ throw new Error(
+ `No mappings would be generated for ${template.name}, possibly due to failed/misconfigured bootstrapping`
+ );
+ }
+ await esClient.indices.putIndexTemplate({ name, body: template });
+ } catch (err) {
+ this.logger.error(`Error updating index template ${name}: ${err.message}`);
+ throw err;
+ }
+ }
+
  public async getExistingLegacyIndexTemplates(
  indexTemplatePattern: string
  ): Promise<estypes.IndicesGetTemplateResponse> {
@@ -335,7 +357,7 @@ export class ClusterClientAdapter<TDoc extends { body: AliasAny; index: string }
  }
  }
 
- public async createDataStream(name: string, body: Record<string, unknown> = {}): Promise<void> {
+ public async createDataStream(name: string): Promise<void> {
  this.logger.info(`Creating datastream ${name}`);
  try {
  const esClient = await this.elasticsearchClientPromise;
@@ -347,6 +369,23 @@ export class ClusterClientAdapter<TDoc extends { body: AliasAny; index: string }
  }
  }
 
+ public async updateConcreteIndices(name: string): Promise<void> {
+ this.logger.info(`Updating concrete index mappings for ${name}`);
+ try {
+ const esClient = await this.elasticsearchClientPromise;
+ const simulatedIndexMapping = await esClient.indices.simulateIndexTemplate({ name });
+ const simulatedMapping = get(simulatedIndexMapping, ['template', 'mappings']);
+
+ if (simulatedMapping != null) {
+ await esClient.indices.putMapping({ index: name, body: simulatedMapping });
+ this.logger.debug(`Successfully updated concrete index mappings for ${name}`);
+ }
+ } catch (err) {
+ this.logger.error(`Error updating index mappings for ${name}: ${err.message}`);
+ throw err;
+ }
+ }
+
  public async queryEventsBySavedObjects(
  queryOptions: FindEventsOptionsBySavedObjectFilter
  ): Promise<QueryEventsBySavedObjectResult> {

x-pack/plugins/event_log/server/es/init.test.ts

@@ -18,7 +18,7 @@ describe('initializeEs', () => {
  esContext.esAdapter.getExistingIndexAliases.mockResolvedValue({});
  });
 
- test(`should update existing index templates if any exist and are not hidden`, async () => {
+ test(`should update existing index templates to hidden if any exist and are not hidden`, async () => {
  const testTemplate = {
  order: 0,
  index_patterns: ['foo-bar-*'],
@@ -393,14 +393,16 @@ describe('initializeEs', () => {
  await initializeEs(esContext);
  expect(esContext.esAdapter.doesIndexTemplateExist).toHaveBeenCalled();
  expect(esContext.esAdapter.createIndexTemplate).toHaveBeenCalled();
+ expect(esContext.esAdapter.updateIndexTemplate).not.toHaveBeenCalled();
  });
 
- test(`shouldn't create index template if it already exists`, async () => {
+ test(`should update index template if it already exists`, async () => {
  esContext.esAdapter.doesIndexTemplateExist.mockResolvedValue(true);
 
  await initializeEs(esContext);
  expect(esContext.esAdapter.doesIndexTemplateExist).toHaveBeenCalled();
  expect(esContext.esAdapter.createIndexTemplate).not.toHaveBeenCalled();
+ expect(esContext.esAdapter.updateIndexTemplate).toHaveBeenCalled();
  });
 
  test(`should create data stream if it doesn't exist`, async () => {
@@ -409,14 +411,16 @@ describe('initializeEs', () => {
  await initializeEs(esContext);
  expect(esContext.esAdapter.doesDataStreamExist).toHaveBeenCalled();
  expect(esContext.esAdapter.createDataStream).toHaveBeenCalled();
+ expect(esContext.esAdapter.updateConcreteIndices).not.toHaveBeenCalled();
  });
 
- test(`shouldn't create data stream if it already exists`, async () => {
+ test(`should update indices of data stream if it already exists`, async () => {
  esContext.esAdapter.doesDataStreamExist.mockResolvedValue(true);
 
  await initializeEs(esContext);
  expect(esContext.esAdapter.doesDataStreamExist).toHaveBeenCalled();
  expect(esContext.esAdapter.createDataStream).not.toHaveBeenCalled();
+ expect(esContext.esAdapter.updateConcreteIndices).toHaveBeenCalled();
  });
 });