[Logs Explorer] Add support for log generation in synthtrace

packages/kbn-apm-synthtrace-client/index.ts

@@ -33,3 +33,4 @@ export { dedot } from './src/lib/utils/dedot';
 export { generateLongId, generateShortId } from './src/lib/utils/generate_id';
 export { appendHash, hashKeysOf } from './src/lib/utils/hash';
 export type { ESDocumentWithOperation, SynthtraceESAction, SynthtraceGenerator } from './src/types';
+export { log, type LogDocument } from './src/lib/logs';

packages/kbn-apm-synthtrace-client/src/lib/logs/index.ts

@@ -0,0 +1,68 @@
+/*
+ * 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 and the Server Side Public License, v 1; you may not use this file except
+ * in compliance with, at your election, the Elastic License 2.0 or the Server
+ * Side Public License, v 1.
+ */
+
+import { Fields } from '../entity';
+import { Serializable } from '../serializable';
+
+export type LogDocument = Fields &
+  Partial<{
+    'input.type': string;
+    'log.file.path'?: string;
+    'service.name': string;
+    'data_stream.namespace': string;
+    'data_stream.type': string;
+    'data_stream.dataset': string;
+    message?: string;
+    'event.dataset': string;
+    'log.level'?: string;
+    'host.name'?: string;
+  }>;
+
+class Log extends Serializable<LogDocument> {
+  service(name: string) {
+    this.fields['service.name'] = name;
+    return this;
+  }
+
+  namespace(value: string) {
+    this.fields['data_stream.namespace'] = value;
+    return this;
+  }
+
+  dataset(value: string) {
+    this.fields['data_stream.dataset'] = value;
+    this.fields['event.dataset'] = value;
+    return this;
+  }
+
+  logLevel(level: string) {
+    this.fields['log.level'] = level;
+    return this;
+  }
+
+  message(message: string) {
+    this.fields.message = message;
+    return this;
+  }
+}
+
+function service(serviceName: string = 'synth-service'): Log {
+  return new Log({
+    'input.type': 'logs',
+    'service.name': serviceName,
+    'data_stream.namespace': 'default',
+    'data_stream.type': 'logs',
+    'data_stream.dataset': 'synth',
+    'event.dataset': 'synth',
+    'host.name': 'synth-host',
+  });
+}
+
+export const log = {
+  service,
+};

packages/kbn-apm-synthtrace/index.ts

@@ -17,6 +17,8 @@ export { AssetsSynthtraceEsClient } from './src/lib/assets/assets_synthtrace_es_
 
 export { MonitoringSynthtraceEsClient } from './src/lib/monitoring/monitoring_synthtrace_es_client';
 
+export { LogsSynthtraceEsClient } from './src/lib/logs/logs_synthtrace_es_client';
+
 export {
   addObserverVersionTransform,
   deleteSummaryFieldTransform,

packages/kbn-apm-synthtrace/src/cli/run_synthtrace.ts

@@ -66,6 +66,12 @@ function options(y: Argv) {
         return arg as Record<string, any> | undefined;
       },
     })
+    .option('type', {
+      describe:
+        'Type of data to be generated, defaults to APM events, can be logs for generating logs',
+      default: 'apm',
+      string: true,
+    })
     .showHelpOnFail(false);
 }
 

packages/kbn-apm-synthtrace/src/cli/utils/bootstrap.ts

@@ -8,6 +8,7 @@
 
 import { createLogger } from '../../lib/utils/create_logger';
 import { getEsClient } from './get_es_client';
+import { getLogsEsClient } from './get_logs_es_client';
 import { getKibanaClient } from './get_kibana_client';
 import { getServiceUrls } from './get_service_urls';
 import { RunOptions } from './parse_run_cli_flags';
@@ -33,15 +34,23 @@ export async function bootstrap(runOptions: RunOptions) {
     version,
   });
 
+  const logsEsClient = getLogsEsClient({
+    target: esUrl,
+    logger,
+    concurrency: runOptions.concurrency,
+  });
+
   await kibanaClient.installApmPackage(latestPackageVersion);
 
   if (runOptions.clean) {
     await apmEsClient.clean();
+    await logsEsClient.clean();
   }
 
   return {
     logger,
     apmEsClient,
+    logsEsClient,
     version,
     kibanaUrl,
     esUrl,

packages/kbn-apm-synthtrace/src/cli/utils/get_logs_es_client.ts

@@ -0,0 +1,31 @@
+/*
+ * 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 and the Server Side Public License, v 1; you may not use this file except
+ * in compliance with, at your election, the Elastic License 2.0 or the Server
+ * Side Public License, v 1.
+ */
+
+import { Client } from '@elastic/elasticsearch';
+import { LogsSynthtraceEsClient } from '../../lib/logs/logs_synthtrace_es_client';
+import { Logger } from '../../lib/utils/create_logger';
+import { RunOptions } from './parse_run_cli_flags';
+
+export function getLogsEsClient({
+  target,
+  logger,
+  concurrency,
+}: Pick<RunOptions, 'concurrency'> & {
+  target: string;
+  logger: Logger;
+}) {
+  const client = new Client({
+    node: target,
+  });
+
+  return new LogsSynthtraceEsClient({
+    client,
+    logger,
+    concurrency,
+  });
+}

packages/kbn-apm-synthtrace/src/cli/utils/parse_run_cli_flags.ts

@@ -69,7 +69,8 @@ export function parseRunCliFlags(flags: RunCliFlags) {
       'kibana',
       'concurrency',
       'versionOverride',
-      'clean'
+      'clean',
+      'type'
     ),
     logLevel: parsedLogLevel,
     file: parsedFile,

packages/kbn-apm-synthtrace/src/cli/utils/start_live_data_upload.ts

@@ -10,6 +10,7 @@ import { timerange } from '@kbn/apm-synthtrace-client';
 import { castArray } from 'lodash';
 import { PassThrough, Readable, Writable } from 'stream';
 import { isGeneratorObject } from 'util/types';
+import { ApmSynthtraceEsClient, LogsSynthtraceEsClient } from '../../..';
 import { awaitStream } from '../../lib/utils/wait_until_stream_finished';
 import { bootstrap } from './bootstrap';
 import { getScenario } from './get_scenario';
@@ -23,9 +24,14 @@ export async function startLiveDataUpload({
   start: Date;
 }) {
   const file = runOptions.file;
+  const clientType = runOptions.type;
 
-  const { logger, apmEsClient } = await bootstrap(runOptions);
+  const { logger, apmEsClient, logsEsClient } = await bootstrap(runOptions);
 
+  let client: ApmSynthtraceEsClient | LogsSynthtraceEsClient = apmEsClient;
+  if (clientType === 'log') {
+    client = logsEsClient;
+  }
   const scenario = await getScenario({ file, logger });
   const { generate } = await scenario({ ...runOptions, logger });
 
@@ -36,7 +42,7 @@ export async function startLiveDataUpload({
     objectMode: true,
   });
 
-  apmEsClient.index(stream);
+  client.index(stream);
 
   function closeStream() {
     stream.end(() => {
@@ -74,7 +80,7 @@ export async function startLiveDataUpload({
 
       await awaitStream(concatenatedStream);
 
-      await apmEsClient.refresh();
+      await client.refresh();
 
       requestedUntil = bucketTo;
     }
@@ -85,6 +91,7 @@ export async function startLiveDataUpload({
     await delay(bucketSizeInMs);
   } while (true);
 }
+
 async function delay(ms: number) {
   return await new Promise((resolve) => setTimeout(resolve, ms));
 }

packages/kbn-apm-synthtrace/src/cli/utils/synthtrace_worker.ts

@@ -13,6 +13,8 @@ import { getEsClient } from './get_es_client';
 import { getScenario } from './get_scenario';
 import { loggerProxy } from './logger_proxy';
 import { RunOptions } from './parse_run_cli_flags';
+import { ApmSynthtraceEsClient, LogsSynthtraceEsClient } from '../../..';
+import { getLogsEsClient } from './get_logs_es_client';
 
 export interface WorkerData {
   bucketFrom: Date;
@@ -34,8 +36,21 @@ async function start() {
     version,
   });
 
+  const logsEsClient = getLogsEsClient({
+    concurrency: runOptions.concurrency,
+    target: esUrl,
+    logger,
+  });
+
+  const clientType = runOptions.type;
+
   const file = runOptions.file;
 
+  let client: ApmSynthtraceEsClient | LogsSynthtraceEsClient = apmEsClient;
+  if (clientType === 'log') {
+    client = logsEsClient;
+  }
+
   const scenario = await logger.perf('get_scenario', () => getScenario({ file, logger }));
 
   logger.info(`Running scenario from ${bucketFrom.toISOString()} to ${bucketTo.toISOString()}`);
@@ -65,8 +80,8 @@ async function start() {
   }, 5000);
 
   await logger.perf('index_scenario', async () => {
-    await apmEsClient.index(generators);
-    await apmEsClient.refresh();
+    await client.index(generators);
+    await client.refresh();
   });
 }
 

packages/kbn-apm-synthtrace/src/lib/logs/logs_synthtrace_es_client.ts

@@ -0,0 +1,75 @@
+/*
+ * 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 and the Server Side Public License, v 1; you may not use this file except
+ * in compliance with, at your election, the Elastic License 2.0 or the Server
+ * Side Public License, v 1.
+ */
+
+import { Client } from '@elastic/elasticsearch';
+import { dedot, ESDocumentWithOperation } from '@kbn/apm-synthtrace-client';
+import { pipeline, Readable, Transform } from 'stream';
+import { LogDocument } from '@kbn/apm-synthtrace-client/src/lib/logs';
+import { SynthtraceEsClient, SynthtraceEsClientOptions } from '../shared/base_client';
+import { getSerializeTransform } from '../shared/get_serialize_transform';
+import { Logger } from '../utils/create_logger';
+
+export type LogsSynthtraceEsClientOptions = Omit<SynthtraceEsClientOptions, 'pipeline'>;
+
+export class LogsSynthtraceEsClient extends SynthtraceEsClient<LogDocument> {
+  constructor(options: { client: Client; logger: Logger } & LogsSynthtraceEsClientOptions) {
+    super({
+      ...options,
+      pipeline: logsPipeline(),
+    });
+    this.dataStreams = ['logs-*'];
+  }
+}
+
+function logsPipeline() {
+  return (base: Readable) => {
+    return pipeline(
+      base,
+      getSerializeTransform<LogDocument>(),
+      getRoutingTransform(),
+      getDedotTransform(),
+      (err: unknown) => {
+        if (err) {
+          throw err;
+        }
+      }
+    );
+  };
+}
+
+function getRoutingTransform() {
+  return new Transform({
+    objectMode: true,
+    transform(document: ESDocumentWithOperation<LogDocument>, encoding, callback) {
+      if (
+        'data_stream.type' in document &&
+        'data_stream.dataset' in document &&
+        'data_stream.namespace' in document
+      ) {
+        document._index = `${document['data_stream.type']}-${document['data_stream.dataset']}-${document['data_stream.namespace']}`;
+      } else {
+        throw new Error('Cannot determine index for event');
+      }
+
+      callback(null, document);
+    },
+  });
+}
+
+function getDedotTransform() {
+  return new Transform({
+    objectMode: true,
+    transform(document: LogDocument, encoding, callback) {
+      const target: Record<string, any> = dedot(document, {});
+      delete target.meta;
+      target['@timestamp'] = new Date(target['@timestamp']!).toISOString();
+
+      callback(null, target);
+    },
+  });
+}

packages/kbn-apm-synthtrace/src/lib/shared/get_serialize_transform.ts

@@ -9,13 +9,13 @@
 import { ApmFields, Serializable } from '@kbn/apm-synthtrace-client';
 import { Transform } from 'stream';
 
-export function getSerializeTransform() {
-  const buffer: ApmFields[] = [];
+export function getSerializeTransform<TFields = ApmFields>() {
+  const buffer: TFields[] = [];
 
   let cb: (() => void) | undefined;
 
-  function push(stream: Transform, events: ApmFields[], callback?: () => void) {
-    let event: ApmFields | undefined;
+  function push(stream: Transform, events: TFields[], callback?: () => void) {
+    let event: TFields | undefined;
     while ((event = events.shift())) {
       if (!stream.push(event)) {
         buffer.push(...events);
@@ -37,7 +37,7 @@ export function getSerializeTransform() {
         push(this, nextEvents, nextCallback);
       }
     },
-    write(chunk: Serializable<ApmFields>, encoding, callback) {
+    write(chunk: Serializable<TFields>, encoding, callback) {
       push(this, chunk.serialize(), callback);
     },
   });