[Synthtrace] Support Non-ECS Logs

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

@@ -33,5 +33,5 @@ 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';
+export { log, type LogDocument, LONG_FIELD_NAME } from './src/lib/logs';
 export { type AssetDocument } from './src/lib/assets';

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

@@ -10,6 +10,9 @@ import { randomInt } from 'crypto';
 import { Fields } from '../entity';
 import { Serializable } from '../serializable';
 
+export const LONG_FIELD_NAME =
+  'thisisaverylongfieldnamethatevendoesnotcontainanyspaceswhyitcouldpotentiallybreakouruiinseveralplaces';
+
 const LOGSDB_DATASET_PREFIX = 'logsdb.';
 
 interface LogsOptions {
@@ -63,6 +66,12 @@ export type LogDocument = Fields &
     'event.duration': number;
     'event.start': Date;
     'event.end': Date;
+    date: Date;
+    severity: string;
+    msg: string;
+    svc: string;
+    hostname: string;
+    [LONG_FIELD_NAME]: string;
   }>;
 
 class Log extends Serializable<LogDocument> {
@@ -123,6 +132,11 @@ class Log extends Serializable<LogDocument> {
     super.timestamp(time);
     return this;
   }
+
+  deleteField(fieldName: keyof LogDocument) {
+    delete this.fields[fieldName];
+    return this;
+  }
 }
 
 function create(logsOptions: LogsOptions = defaultLogsOptions): Log {

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

@@ -0,0 +1,17 @@
+/*
+ * 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 type { MappingTypeMapping } from '@elastic/elasticsearch/lib/api/types';
+
+export const timestampDateMapping: MappingTypeMapping = {
+  properties: {
+    '@timestamp': {
+      type: 'date',
+    },
+  },
+};

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

@@ -10,6 +10,7 @@ import { Client } from '@elastic/elasticsearch';
 import { ESDocumentWithOperation } from '@kbn/apm-synthtrace-client';
 import { pipeline, Readable, Transform } from 'stream';
 import { LogDocument } from '@kbn/apm-synthtrace-client/src/lib/logs';
+import { MappingTypeMapping } from '@elastic/elasticsearch/lib/api/types';
 import { SynthtraceEsClient, SynthtraceEsClientOptions } from '../shared/base_client';
 import { getSerializeTransform } from '../shared/get_serialize_transform';
 import { Logger } from '../utils/create_logger';
@@ -24,6 +25,7 @@ export class LogsSynthtraceEsClient extends SynthtraceEsClient<LogDocument> {
       pipeline: logsPipeline(),
     });
     this.dataStreams = ['logs-*-*'];
+    this.indices = ['cloud-logs-*-*'];
   }
 
   async createIndexTemplate(name: IndexTemplateName) {
@@ -40,6 +42,23 @@ export class LogsSynthtraceEsClient extends SynthtraceEsClient<LogDocument> {
       this.logger.error(`Index template creation failed: ${name} - ${err.message}`);
     }
   }
+
+  async createIndex(index: string, mappings?: MappingTypeMapping) {
+    try {
+      const isIndexExisting = await this.client.indices.exists({ index });
+
+      if (isIndexExisting) {
+        this.logger.info(`Index already exists: ${index}`);
+        return;
+      }
+
+      await this.client.indices.create({ index, mappings });
+
+      this.logger.info(`Index successfully created: ${index}`);
+    } catch (err) {
+      this.logger.error(`Index creation failed: ${index} - ${err.message}`);
+    }
+  }
 }
 
 function logsPipeline() {

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

@@ -53,6 +53,17 @@ export class SynthtraceEsClient<TFields extends Fields> {
       )}"`
     );
 
+    const resolvedIndices = this.indices.length
+      ? (
+          await this.client.indices.resolveIndex({
+            name: this.indices.join(','),
+            expand_wildcards: ['open', 'hidden'],
+            // @ts-expect-error ignore_unavailable is not in the type definition, but it is accepted by es
+            ignore_unavailable: true,
+          })
+        ).indices.map((index: { name: string }) => index.name)
+      : [];
+
     await Promise.all([
       ...(this.dataStreams.length
         ? [
@@ -62,10 +73,10 @@ export class SynthtraceEsClient<TFields extends Fields> {
             }),
           ]
         : []),
-      ...(this.indices.length
+      ...(resolvedIndices.length
         ? [
             this.client.indices.delete({
-              index: this.indices.join(','),
+              index: resolvedIndices.join(','),
               expand_wildcards: ['open', 'hidden'],
               ignore_unavailable: true,
               allow_no_indices: true,

packages/kbn-apm-synthtrace/src/scenarios/simple_non_ecs_logs.ts

@@ -0,0 +1,161 @@
+/*
+ * 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 {
+  LogDocument,
+  log,
+  generateShortId,
+  generateLongId,
+  LONG_FIELD_NAME,
+} from '@kbn/apm-synthtrace-client';
+import moment from 'moment';
+import { Scenario } from '../cli/scenario';
+import { IndexTemplateName } from '../lib/logs/custom_logsdb_index_templates';
+import { withClient } from '../lib/utils/with_client';
+import {
+  getServiceName,
+  getCluster,
+  getCloudProvider,
+  getCloudRegion,
+} from './helpers/logs_mock_data';
+import { parseLogsScenarioOpts } from './helpers/logs_scenario_opts_parser';
+import { timestampDateMapping } from '../lib/logs/custom_logsdb_indices';
+
+// Logs Data logic
+const MESSAGE_LOG_LEVELS = [
+  { message: 'A simple log with something random <random> in the middle', level: 'info' },
+  { message: 'Yet another debug log', level: 'debug' },
+  { message: 'Error with certificate: "ca_trusted_fingerprint"', level: 'error' },
+];
+
+const scenario: Scenario<LogDocument> = async (runOptions) => {
+  const { isLogsDb } = parseLogsScenarioOpts(runOptions.scenarioOpts);
+
+  const constructLogsCommonData = () => {
+    const index = Math.floor(Math.random() * 3);
+    const serviceName = getServiceName(index);
+    const { message, level } = MESSAGE_LOG_LEVELS[index];
+    const { clusterId, clusterName, namespace } = getCluster(index);
+    const cloudRegion = getCloudRegion(index);
+
+    const commonLongEntryFields: LogDocument = {
+      'trace.id': generateShortId(),
+      'agent.name': 'nodejs',
+      'orchestrator.cluster.name': clusterName,
+      'orchestrator.cluster.id': clusterId,
+      'orchestrator.namespace': namespace,
+      'container.name': `${serviceName}-${generateShortId()}`,
+      'orchestrator.resource.id': generateShortId(),
+      'cloud.provider': getCloudProvider(),
+      'cloud.region': cloudRegion,
+      'cloud.availability_zone': `${cloudRegion}a`,
+      'cloud.project.id': generateShortId(),
+      'cloud.instance.id': generateShortId(),
+      'log.file.path': `/logs/${generateLongId()}/error.txt`,
+      severity: level,
+      svc: serviceName,
+      msg: message.replace('<random>', generateShortId()),
+      [LONG_FIELD_NAME]: 'test',
+    };
+
+    return {
+      index,
+      serviceName,
+      cloudRegion,
+      commonLongEntryFields,
+    };
+  };
+
+  return {
+    bootstrap: async ({ logsEsClient }) => {
+      await logsEsClient.createIndex('cloud-logs-synth.1-default', timestampDateMapping);
+      await logsEsClient.createIndex('cloud-logs-synth.2-default');
+      if (isLogsDb) await logsEsClient.createIndexTemplate(IndexTemplateName.LogsDb);
+    },
+    generate: ({ range, clients: { logsEsClient } }) => {
+      const { logger } = runOptions;
+
+      const logsWithNonEcsFields = range
+        .interval('1m')
+        .rate(1)
+        .generator((timestamp) => {
+          return Array(3)
+            .fill(0)
+            .map(() => {
+              const { commonLongEntryFields } = constructLogsCommonData();
+
+              return log
+                .create({ isLogsDb })
+                .deleteField('host.name')
+                .defaults({
+                  ...commonLongEntryFields,
+                  hostname: 'synth-host',
+                })
+                .dataset('custom.synth')
+                .timestamp(timestamp);
+            });
+        });
+
+      const logsOutsideDsnsWithTimestamp = range
+        .interval('1m')
+        .rate(1)
+        .generator((timestamp) => {
+          return Array(3)
+            .fill(0)
+            .map(() => {
+              const { commonLongEntryFields } = constructLogsCommonData();
+
+              return log
+                .create({ isLogsDb })
+                .deleteField('host.name')
+                .deleteField('data_stream.type')
+                .defaults({
+                  ...commonLongEntryFields,
+                  'data_stream.type': 'cloud-logs',
+                  hostname: 'synth-host1',
+                })
+                .dataset('synth.1')
+                .timestamp(timestamp);
+            });
+        });
+
+      const logsOutsideDsnsWithoutTimestamp = range
+        .interval('1m')
+        .rate(1)
+        .generator((timestamp) => {
+          return Array(3)
+            .fill(0)
+            .map(() => {
+              const { commonLongEntryFields } = constructLogsCommonData();
+
+              return log
+                .create({ isLogsDb })
+                .deleteField('host.name')
+                .deleteField('data_stream.type')
+                .defaults({
+                  ...commonLongEntryFields,
+                  hostname: 'synth-host2',
+                  'data_stream.type': 'cloud-logs',
+                  date: moment(timestamp).toDate(),
+                })
+                .dataset('synth.2');
+            });
+        });
+
+      return withClient(
+        logsEsClient,
+        logger.perf('generating_logs', () => [
+          logsWithNonEcsFields,
+          logsOutsideDsnsWithTimestamp,
+          logsOutsideDsnsWithoutTimestamp,
+        ])
+      );
+    },
+  };
+};
+
+export default scenario;