[event_log] index event docs in bulk instead of individually

resolves #55634

Buffers event docs being written for a fixed interval / buffer size,
and indexes those docs via a bulk ES call.

Author: pmuellr

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

@@ -9,13 +9,15 @@ import { IClusterClientAdapter } from './cluster_client_adapter';
 const createClusterClientMock = () => {
   const mock: jest.Mocked<IClusterClientAdapter> = {
     indexDocument: jest.fn(),
+    indexDocuments: jest.fn(),
     doesIlmPolicyExist: jest.fn(),
     createIlmPolicy: jest.fn(),
     doesIndexTemplateExist: jest.fn(),
     createIndexTemplate: jest.fn(),
     doesAliasExist: jest.fn(),
     createIndex: jest.fn(),
     queryEventsBySavedObject: jest.fn(),
+    shutdown: jest.fn(),
   };
   return mock;
 };

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

@@ -4,14 +4,21 @@
  * you may not use this file except in compliance with the Elastic License.
  */
 
-import { LegacyClusterClient, Logger } from 'src/core/server';
+import { LegacyClusterClient } from 'src/core/server';
 import { elasticsearchServiceMock, loggingSystemMock } from 'src/core/server/mocks';
-import { ClusterClientAdapter, IClusterClientAdapter } from './cluster_client_adapter';
+import {
+  ClusterClientAdapter,
+  IClusterClientAdapter,
+  EVENT_BUFFER_LENGTH,
+} from './cluster_client_adapter';
+import { contextMock } from './context.mock';
 import { findOptionsSchema } from '../event_log_client';
+import { delay } from '../lib/delay';
 
 type EsClusterClient = Pick<jest.Mocked<LegacyClusterClient>, 'callAsInternalUser' | 'asScoped'>;
+type MockedLogger = ReturnType<typeof loggingSystemMock['createLogger']>;
 
-let logger: Logger;
+let logger: MockedLogger;
 let clusterClient: EsClusterClient;
 let clusterClientAdapter: IClusterClientAdapter;
 
@@ -21,22 +28,97 @@ beforeEach(() => {
   clusterClientAdapter = new ClusterClientAdapter({
     logger,
     clusterClientPromise: Promise.resolve(clusterClient),
+    context: contextMock.create(),
   });
 });
 
 describe('indexDocument', () => {
-  test('should call cluster client with given doc', async () => {
-    await clusterClientAdapter.indexDocument({ args: true });
-    expect(clusterClient.callAsInternalUser).toHaveBeenCalledWith('index', {
-      args: true,
+  test('should call cluster client bulk with given doc', async () => {
+    clusterClientAdapter.indexDocument({ body: { message: 'foo' }, index: 'event-log' });
+
+    await retryUntil('cluster client bulk called', () => {
+      return clusterClient.callAsInternalUser.mock.calls.length !== 0;
+    });
+
+    expect(clusterClient.callAsInternalUser).toHaveBeenCalledWith('bulk', {
+      body: [{ create: { _index: 'event-log' } }, { message: 'foo' }],
     });
   });
 
-  test('should throw error when cluster client throws an error', async () => {
-    clusterClient.callAsInternalUser.mockRejectedValue(new Error('Fail'));
-    await expect(
-      clusterClientAdapter.indexDocument({ args: true })
-    ).rejects.toThrowErrorMatchingInlineSnapshot(`"Fail"`);
+  test('should log an error when cluster client throws an error', async () => {
+    clusterClient.callAsInternalUser.mockRejectedValue(new Error('expected failure'));
+    clusterClientAdapter.indexDocument({ body: { message: 'foo' }, index: 'event-log' });
+    await retryUntil('cluster client bulk called', () => {
+      return logger.error.mock.calls.length !== 0;
+    });
+
+    const expectedMessage = `error writing bulk events: "expected failure"; docs: [{"create":{"_index":"event-log"}},{"message":"foo"}]`;
+    expect(logger.error).toHaveBeenCalledWith(expectedMessage);
+  });
+});
+
+describe('shutdown()', () => {
+  test('should work if no docs have been written', async () => {
+    const result = await clusterClientAdapter.shutdown();
+    expect(result).toBeFalsy();
+  });
+
+  test('should work if some docs have been written', async () => {
+    clusterClientAdapter.indexDocument({ body: { message: 'foo' }, index: 'event-log' });
+    const resultPromise = clusterClientAdapter.shutdown();
+
+    await retryUntil('cluster client bulk called', () => {
+      return clusterClient.callAsInternalUser.mock.calls.length !== 0;
+    });
+
+    const result = await resultPromise;
+    expect(result).toBeFalsy();
+  });
+});
+
+describe('buffering documents', () => {
+  test('should write buffered docs after timeout', async () => {
+    // write EVENT_BUFFER_LENGTH - 1 docs
+    for (let i = 0; i < EVENT_BUFFER_LENGTH - 1; i++) {
+      clusterClientAdapter.indexDocument({ body: { message: `foo ${i}` }, index: 'event-log' });
+    }
+
+    await retryUntil('cluster client bulk called', () => {
+      return clusterClient.callAsInternalUser.mock.calls.length !== 0;
+    });
+
+    const expectedBody = [];
+    for (let i = 0; i < EVENT_BUFFER_LENGTH - 1; i++) {
+      expectedBody.push({ create: { _index: 'event-log' } }, { message: `foo ${i}` });
+    }
+
+    expect(clusterClient.callAsInternalUser).toHaveBeenCalledWith('bulk', {
+      body: expectedBody,
+    });
+  });
+
+  test('should write buffered docs after buffer exceeded', async () => {
+    // write EVENT_BUFFER_LENGTH + 1 docs
+    for (let i = 0; i < EVENT_BUFFER_LENGTH + 1; i++) {
+      clusterClientAdapter.indexDocument({ body: { message: `foo ${i}` }, index: 'event-log' });
+    }
+
+    await retryUntil('cluster client bulk called', () => {
+      return clusterClient.callAsInternalUser.mock.calls.length >= 2;
+    });
+
+    const expectedBody = [];
+    for (let i = 0; i < EVENT_BUFFER_LENGTH; i++) {
+      expectedBody.push({ create: { _index: 'event-log' } }, { message: `foo ${i}` });
+    }
+
+    expect(clusterClient.callAsInternalUser).toHaveBeenNthCalledWith(1, 'bulk', {
+      body: expectedBody,
+    });
+
+    expect(clusterClient.callAsInternalUser).toHaveBeenNthCalledWith(2, 'bulk', {
+      body: [{ create: { _index: 'event-log' } }, { message: `foo 100` }],
+    });
   });
 });
 
@@ -575,3 +657,29 @@ describe('queryEventsBySavedObject', () => {
     `);
   });
 });
+
+type RetryableFunction = () => boolean;
+
+const RETRY_UNTIL_DEFAULT_COUNT = 20;
+const RETRY_UNTIL_DEFAULT_WAIT = 1000; // milliseconds
+
+async function retryUntil(
+  label: string,
+  fn: RetryableFunction,
+  count: number = RETRY_UNTIL_DEFAULT_COUNT,
+  wait: number = RETRY_UNTIL_DEFAULT_WAIT
+): Promise<boolean> {
+  while (count > 0) {
+    count--;
+
+    if (fn()) return true;
+
+    // eslint-disable-next-line no-console
+    console.log(`attempt failed waiting for "${label}", attempts left: ${count}`);
+
+    if (count === 0) return false;
+    await delay(wait);
+  }
+
+  return false;
+}

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

@@ -4,20 +4,32 @@
  * you may not use this file except in compliance with the Elastic License.
  */
 
+import { Subject, Subscription } from 'rxjs';
+import { bufferTime, filter, switchMap } from 'rxjs/operators';
 import { reject, isUndefined } from 'lodash';
 import { SearchResponse, Client } from 'elasticsearch';
 import type { PublicMethodsOf } from '@kbn/utility-types';
 import { Logger, LegacyClusterClient } from 'src/core/server';
-
-import { IValidatedEvent, SAVED_OBJECT_REL_PRIMARY } from '../types';
+import { ReadySignal, createReadySignal } from '../lib/ready_signal';
+import { EsContext } from '.';
+import { IEvent, IValidatedEvent, SAVED_OBJECT_REL_PRIMARY } from '../types';
 import { FindOptionsType } from '../event_log_client';
 
+export const EVENT_BUFFER_TIME = 1000; // milliseconds
+export const EVENT_BUFFER_LENGTH = 100;
+
 export type EsClusterClient = Pick<LegacyClusterClient, 'callAsInternalUser' | 'asScoped'>;
 export type IClusterClientAdapter = PublicMethodsOf<ClusterClientAdapter>;
 
+export interface Doc {
+  index: string;
+  body: IEvent;
+}
+
 export interface ConstructorOpts {
   logger: Logger;
   clusterClientPromise: Promise<EsClusterClient>;
+  context: EsContext;
 }
 
 export interface QueryEventsBySavedObjectResult {
@@ -30,14 +42,71 @@ export interface QueryEventsBySavedObjectResult {
 export class ClusterClientAdapter {
   private readonly logger: Logger;
   private readonly clusterClientPromise: Promise<EsClusterClient>;
+  private readonly docBuffer$: Subject<Doc>;
+  private readonly docsBufferedSubscription: Subscription;
+  private readonly doneWriting: ReadySignal;
+  private readonly context: EsContext;
 
   constructor(opts: ConstructorOpts) {
     this.logger = opts.logger;
     this.clusterClientPromise = opts.clusterClientPromise;
+    this.context = opts.context;
+    this.docBuffer$ = new Subject<Doc>();
+    this.doneWriting = createReadySignal();
+
+    // buffer event log docs for time / buffer length, ignore empty
+    // buffers, then index the buffered docs
+    const docsBuffered$ = this.docBuffer$.pipe(
+      bufferTime(EVENT_BUFFER_TIME, null, EVENT_BUFFER_LENGTH),
+      filter((docs) => docs.length > 0),
+      switchMap(async (docs) => await this.indexDocuments(docs))
+    );
+
+    // kick everything off, when the stream closes, signal finished
+    const { doneWriting } = this;
+    this.docsBufferedSubscription = docsBuffered$.subscribe({
+      complete() {
+        doneWriting.signal();
+      },
+    });
+  }
+
+  // This will be called at plugin stop() time; the assumption is any plugins
+  // depending on the event_log will already be stopped, and so will not be
+  // writing more event docs.  So we shut down the docBuffer$ observable,
+  // and wait for the docsBufffered$ observable to complete
+  public async shutdown(): Promise<void> {
+    this.docBuffer$.complete();
+    await this.doneWriting.wait();
+    this.docsBufferedSubscription.unsubscribe();
+  }
+
+  public indexDocument(doc: Doc): void {
+    this.docBuffer$.next(doc);
   }
 
-  public async indexDocument(doc: unknown): Promise<void> {
-    await this.callEs<ReturnType<Client['index']>>('index', doc);
+  async indexDocuments(docs: Doc[]): Promise<void> {
+    // if es initialization failed, don't try to index
+    if (!(await this.context.waitTillReady())) {
+      return;
+    }
+
+    const bulkBody: Array<Record<string, unknown>> = [];
+
+    for (const doc of docs) {
+      if (doc.body === undefined) continue;
+
+      bulkBody.push({ create: { _index: doc.index } });
+      bulkBody.push(doc.body);
+    }
+
+    try {
+      await this.callEs<ReturnType<Client['bulk']>>('bulk', { body: bulkBody });
+    } catch (err) {
+      this.logger.error(
+        `error writing bulk events: "${err.message}"; docs: ${JSON.stringify(bulkBody)}`
+      );
+    }
   }
 
   public async doesIlmPolicyExist(policyName: string): Promise<boolean> {

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

@@ -18,6 +18,7 @@ const createContextMock = () => {
     logger: loggingSystemMock.createLogger(),
     esNames: namesMock.create(),
     initialize: jest.fn(),
+    shutdown: jest.fn(),
     waitTillReady: jest.fn(async () => true),
     esAdapter: clusterClientAdapterMock.create(),
     initialized: true,

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

@@ -18,6 +18,7 @@ export interface EsContext {
   esNames: EsNames;
   esAdapter: IClusterClientAdapter;
   initialize(): void;
+  shutdown(): Promise<void>;
   waitTillReady(): Promise<boolean>;
   initialized: boolean;
 }
@@ -52,6 +53,7 @@ class EsContextImpl implements EsContext {
     this.esAdapter = new ClusterClientAdapter({
       logger: params.logger,
       clusterClientPromise: params.clusterClientPromise,
+      context: this,
     });
   }
 
@@ -74,6 +76,10 @@ class EsContextImpl implements EsContext {
     });
   }
 
+  async shutdown() {
+    await this.esAdapter.shutdown();
+  }
+
   // waits till the ES initialization is done, returns true if it was successful,
   // false if it was not successful
   async waitTillReady(): Promise<boolean> {

x-pack/plugins/event_log/server/event_logger.test.ts

@@ -59,7 +59,8 @@ describe('EventLogger', () => {
     eventLogger.logEvent({});
     await waitForLogEvent(systemLogger);
     delay(WRITE_LOG_WAIT_MILLIS); // sleep a bit longer since event logging is async
-    expect(esContext.esAdapter.indexDocument).not.toHaveBeenCalled();
+    expect(esContext.esAdapter.indexDocument).toHaveBeenCalled();
+    expect(esContext.esAdapter.indexDocuments).not.toHaveBeenCalled();
   });
 
   test('method logEvent() writes expected default values', async () => {

x-pack/plugins/event_log/server/event_logger.ts

@@ -20,14 +20,10 @@ import {
   EventSchema,
 } from './types';
 import { SAVED_OBJECT_REL_PRIMARY } from './types';
+import { Doc } from './es/cluster_client_adapter';
 
 type SystemLogger = Plugin['systemLogger'];
 
-interface Doc {
-  index: string;
-  body: IEvent;
-}
-
 interface IEventLoggerCtorParams {
   esContext: EsContext;
   eventLogService: EventLogService;
@@ -159,44 +155,9 @@ function validateEvent(eventLogService: IEventLogService, event: IEvent): IValid
 export const EVENT_LOGGED_PREFIX = `event logged: `;
 
 function logEventDoc(logger: Logger, doc: Doc): void {
-  setImmediate(() => {
-    logger.info(`${EVENT_LOGGED_PREFIX}${JSON.stringify(doc.body)}`);
-  });
+  logger.info(`event logged: ${JSON.stringify(doc.body)}`);
 }
 
 function indexEventDoc(esContext: EsContext, doc: Doc): void {
-  // TODO:
-  // the setImmediate() on an async function is a little overkill, but,
-  // setImmediate() may be tweakable via node params, whereas async
-  // tweaking is in the v8 params realm, which is very dicey.
-  // Long-term, we should probably create an in-memory queue for this, so
-  // we can explictly see/set the queue lengths.
-
-  // already verified this.clusterClient isn't null above
-  setImmediate(async () => {
-    try {
-      await indexLogEventDoc(esContext, doc);
-    } catch (err) {
-      esContext.logger.warn(`error writing event doc: ${err.message}`);
-      writeLogEventDocOnError(esContext, doc);
-    }
-  });
-}
-
-// whew, the thing that actually writes the event log document!
-async function indexLogEventDoc(esContext: EsContext, doc: unknown) {
-  esContext.logger.debug(`writing to event log: ${JSON.stringify(doc)}`);
-  const success = await esContext.waitTillReady();
-  if (!success) {
-    esContext.logger.debug(`event log did not initialize correctly, event not written`);
-    return;
-  }
-
-  await esContext.esAdapter.indexDocument(doc);
-  esContext.logger.debug(`writing to event log complete`);
-}
-
-// TODO: write log entry to a bounded queue buffer
-function writeLogEventDocOnError(esContext: EsContext, doc: unknown) {
-  esContext.logger.warn(`unable to write event doc: ${JSON.stringify(doc)}`);
+  esContext.esAdapter.indexDocument(doc);
 }