[9.0] [Streams] Replay loghub data with synthtrace (#212120)

.github/CODEOWNERS

@@ -819,6 +819,7 @@ x-pack/platform/packages/shared/kbn-event-stacktrace @elastic/obs-ux-infra_servi
 x-pack/platform/packages/shared/kbn-inference-endpoint-ui-common @elastic/response-ops @elastic/appex-ai-infra @elastic/obs-ai-assistant @elastic/security-generative-ai
 x-pack/platform/packages/shared/kbn-key-value-metadata-table @elastic/obs-ux-infra_services-team @elastic/obs-ux-logs-team
 x-pack/platform/packages/shared/kbn-langchain @elastic/security-generative-ai
+x-pack/platform/packages/shared/kbn-sample-parser @elastic/streams-program-team
 x-pack/platform/packages/shared/kbn-slo-schema @elastic/obs-ux-management-team
 x-pack/platform/packages/shared/logs-overview @elastic/obs-ux-logs-team
 x-pack/platform/packages/shared/ml/aiops_common @elastic/ml-ui

package.json

@@ -1490,6 +1490,7 @@
     "@kbn/repo-source-classifier": "link:packages/kbn-repo-source-classifier",
     "@kbn/repo-source-classifier-cli": "link:packages/kbn-repo-source-classifier-cli",
     "@kbn/reporting-mocks-server": "link:src/platform/packages/private/kbn-reporting/mocks_server",
+    "@kbn/sample-log-parser": "link:x-pack/platform/packages/shared/kbn-sample-parser",
     "@kbn/scout": "link:src/platform/packages/shared/kbn-scout",
     "@kbn/scout-info": "link:src/platform/packages/private/kbn-scout-info",
     "@kbn/scout-oblt": "link:x-pack/solutions/observability/packages/kbn-scout-oblt",

src/platform/packages/private/kbn-journeys/services/synthtrace.ts

@@ -15,7 +15,7 @@ import {
 } from '@kbn/apm-synthtrace';
 import { ToolingLog } from '@kbn/tooling-log';
 import Url from 'url';
-import { Logger } from '@kbn/apm-synthtrace/src/lib/utils/create_logger';
+import { type Logger, extendToolingLog } from '@kbn/apm-synthtrace';
 import { Auth, Es } from '.';
 import { KibanaUrl } from './kibana_url';
 
@@ -39,45 +39,9 @@ export async function getSynthtraceClient(
   }
 }
 
-// Adapting ToolingLog instance to Logger interface
-class LoggerAdapter implements Logger {
-  private log: ToolingLog;
-  private joiner = ', ';
-
-  constructor(log: ToolingLog) {
-    this.log = log;
-  }
-
-  debug(...args: any[]): void {
-    this.log.debug(args.join(this.joiner));
-  }
-
-  info(...args: any[]): void {
-    this.log.info(args.join(this.joiner));
-  }
-
-  warn(...args: any[]): void {
-    this.log.warning(args.join(this.joiner));
-  }
-
-  error(arg: string | Error): void {
-    this.log.error(arg);
-  }
-
-  perf<T>(name: string, cb: () => T): T {
-    const startTime = Date.now();
-    const result = cb();
-    const duration = Date.now() - startTime;
-    const durationInSeconds = duration / 1000;
-    const formattedTime = durationInSeconds.toFixed(3) + 's';
-    this.log.info(`${name} took ${formattedTime}.`);
-    return result;
-  }
-}
-
 async function initInfraSynthtraceClient(options: SynthtraceClientOptions) {
   const { log, es, auth, kbnUrl } = options;
-  const logger: Logger = new LoggerAdapter(log);
+  const logger: Logger = extendToolingLog(log);
 
   const synthKbnClient = new InfraSynthtraceKibanaClient({
     logger,
@@ -99,7 +63,7 @@ async function initInfraSynthtraceClient(options: SynthtraceClientOptions) {
 
 async function initApmSynthtraceClient(options: SynthtraceClientOptions) {
   const { log, es, auth, kbnUrl } = options;
-  const logger: Logger = new LoggerAdapter(log);
+  const logger: Logger = extendToolingLog(log);
   const kibanaUrl = new URL(kbnUrl.get());
   const kibanaUrlWithAuth = Url.format({
     protocol: kibanaUrl.protocol,

src/platform/packages/shared/kbn-apm-synthtrace-client/index.ts

@@ -27,7 +27,7 @@ export { Entity } from './src/lib/entity';
 export { infra, type InfraDocument } from './src/lib/infra';
 export { parseInterval } from './src/lib/interval';
 export { monitoring, type MonitoringDocument } from './src/lib/monitoring';
-export type { Serializable } from './src/lib/serializable';
+export { Serializable } from './src/lib/serializable';
 export { timerange } from './src/lib/timerange';
 export type { Timerange } from './src/lib/timerange';
 export { dedot } from './src/lib/utils/dedot';

src/platform/packages/shared/kbn-apm-synthtrace-client/src/lib/interval.ts

@@ -7,11 +7,13 @@
  * License v3.0 only", or the "Server Side Public License, v 1".
  */
 
+import { ToolingLog } from '@kbn/tooling-log';
 import { castArray } from 'lodash';
 import moment, { unitOfTime } from 'moment';
 import { SynthtraceGenerator } from '../types';
 import { Fields } from './entity';
 import { Serializable } from './serializable';
+import { TimerangeProgressReporter } from './timerange_progress_reporter';
 
 export function parseInterval(interval: string): {
   intervalAmount: number;
@@ -32,6 +34,7 @@ interface IntervalOptions {
   to: Date;
   interval: string;
   rate?: number;
+  log?: ToolingLog;
 }
 
 interface StepDetails {
@@ -86,10 +89,22 @@ export class Interval<TFields extends Fields = Fields> {
     };
 
     let index = 0;
+    const calculateEvery = 10;
+
+    const reporter = this.options.log
+      ? new TimerangeProgressReporter({
+          log: this.options.log,
+          reportEvery: 5000,
+          total: timestamps.length,
+        })
+      : undefined;
 
     for (const timestamp of timestamps) {
       const events = castArray(map(timestamp, index, stepDetails));
       index++;
+      if (index % calculateEvery === 0) {
+        reporter?.next(index);
+      }
       for (const event of events) {
         yield event;
       }

src/platform/packages/shared/kbn-apm-synthtrace-client/src/lib/timerange.ts

@@ -9,15 +9,20 @@
 
 import datemath from '@kbn/datemath';
 import type { Moment } from 'moment';
+import { ToolingLog } from '@kbn/tooling-log';
 import { GaussianEvents } from './gaussian_events';
 import { Interval } from './interval';
 import { PoissonEvents } from './poisson_events';
 
 export class Timerange {
-  constructor(public readonly from: Date, public readonly to: Date) {}
+  constructor(
+    public readonly from: Date,
+    public readonly to: Date,
+    private readonly log?: ToolingLog
+  ) {}
 
   interval(interval: string) {
-    return new Interval({ from: this.from, to: this.to, interval });
+    return new Interval({ from: this.from, to: this.to, interval, log: this.log });
   }
 
   ratePerMinute(rate: number) {
@@ -39,7 +44,7 @@ export class Timerange {
     return Array.from({ length: segmentCount }, (_, i) => {
       const from = new Date(this.from.getTime() + i * segmentDuration);
       const to = new Date(from.getTime() + segmentDuration);
-      return new Timerange(from, to);
+      return new Timerange(from, to, this.log);
     });
   }
 
@@ -65,7 +70,7 @@ function getDateFrom(date: DateLike, now: Date): Date {
   return date.toDate();
 }
 
-export function timerange(from: DateLike, to: DateLike) {
+export function timerange(from: DateLike, to: DateLike, log?: ToolingLog) {
   const now = new Date();
-  return new Timerange(getDateFrom(from, now), getDateFrom(to, now));
+  return new Timerange(getDateFrom(from, now), getDateFrom(to, now), log);
 }

src/platform/packages/shared/kbn-apm-synthtrace-client/src/lib/timerange_progress_reporter.ts

@@ -0,0 +1,82 @@
+/*
+ * 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", the "GNU Affero General Public License v3.0 only", 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", the "GNU Affero General Public
+ * License v3.0 only", or the "Server Side Public License, v 1".
+ */
+
+import { ToolingLog } from '@kbn/tooling-log';
+import { first, last } from 'lodash';
+import moment from 'moment';
+
+interface TimerangeProgressOptions {
+  log: ToolingLog;
+  total: number;
+  reportEvery: number;
+}
+
+export class TimerangeProgressReporter {
+  private readonly startOfRun: number = performance.now();
+
+  private measurements: Array<{
+    measuredAt: number;
+    index: number;
+  }> = [
+    {
+      measuredAt: this.startOfRun,
+      index: 0,
+    },
+  ];
+
+  private lastReported: number = this.startOfRun;
+
+  constructor(private readonly options: TimerangeProgressOptions) {}
+
+  next(index: number) {
+    const now = performance.now();
+
+    this.measurements.unshift({
+      index,
+      measuredAt: now,
+    });
+
+    this.measurements.length = Math.min(10, this.measurements.length);
+
+    const timeSinceLastReported = now - this.lastReported;
+
+    if (timeSinceLastReported >= this.options.reportEvery) {
+      this.report(now);
+    }
+  }
+
+  private report(now: number) {
+    this.lastReported = now;
+
+    const firstMeasurement = first(this.measurements)!;
+    const lastMeasurement = last(this.measurements)!;
+
+    const totalDurationFormatted = moment.duration(now - this.startOfRun).humanize();
+
+    const indicesLeft = this.options.total - lastMeasurement.index;
+
+    const measuredIndicesProcessed = lastMeasurement.index - firstMeasurement.index;
+
+    const measuredDuration = lastMeasurement.measuredAt - firstMeasurement.measuredAt;
+
+    const indicesPerMs = measuredIndicesProcessed / measuredDuration;
+
+    const timeLeft = indicesLeft / indicesPerMs;
+
+    const timeLeftFormatted = moment.duration(timeLeft).humanize(true);
+
+    const totalProgress = lastMeasurement.index / this.options.total;
+
+    this.options.log.info(
+      `progress=${(totalProgress * 100).toPrecision(
+        3
+      )}%, duration=${totalDurationFormatted}, eta=${timeLeftFormatted}`
+    );
+  }
+}

src/platform/packages/shared/kbn-apm-synthtrace-client/tsconfig.json

@@ -15,5 +15,6 @@
   "kbn_references": [
     "@kbn/datemath",
     "@kbn/safer-lodash-set",
+    "@kbn/tooling-log",
   ]
 }

src/platform/packages/shared/kbn-apm-synthtrace/index.ts

@@ -7,7 +7,12 @@
  * License v3.0 only", or the "Server Side Public License, v 1".
  */
 
-export { createLogger, LogLevel } from './src/lib/utils/create_logger';
+export {
+  createLogger,
+  LogLevel,
+  type Logger,
+  extendToolingLog,
+} from './src/lib/utils/create_logger';
 
 export { ApmSynthtraceEsClient } from './src/lib/apm/client/apm_synthtrace_es_client';
 export { ApmSynthtraceKibanaClient } from './src/lib/apm/client/apm_synthtrace_kibana_client';

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

@@ -67,15 +67,55 @@ function options(y: Argv) {
       number: true,
       default: 1,
     })
+    .option('debug', {
+      describe: 'Use a debug log level',
+      boolean: true,
+    })
+    .option('verbose', {
+      describe: 'Use a verbose log level',
+      boolean: true,
+    })
     .option('logLevel', {
       describe: 'Log level',
-      choices: ['trace', 'debug', 'info', 'error'],
+      choices: ['verbose', 'debug', 'info', 'error'],
       default: 'info',
     })
     .option('scenarioOpts', {
       describe: 'Options specific to the scenario',
-      coerce: (arg) => {
-        return arg as Record<string, any> | undefined;
+      type: 'string',
+      coerce: (arg: string): Record<string, unknown> => {
+        if (!arg) {
+          return {};
+        }
+
+        let scenarioOptions: Record<string, unknown> = {};
+
+        try {
+          scenarioOptions = JSON.parse(arg);
+        } catch (error) {
+          scenarioOptions = Object.fromEntries(
+            arg.split(',').map((kv) => {
+              const [key, value] = kv
+                .trim()
+                .split('=')
+                .map((part) => part.trim());
+              if (value === 'true') {
+                return [key, true];
+              }
+              if (value === 'false') {
+                return [key, false];
+              }
+
+              if (!isNaN(Number(value))) {
+                return [key, Number(value)];
+              }
+
+              return [key, value];
+            })
+          );
+        }
+
+        return scenarioOptions;
       },
     })
     .option('assume-package-version', {
@@ -95,20 +135,18 @@ function options(y: Argv) {
 async function run(argv: RunCliFlags) {
   const runOptions = parseRunCliFlags(argv);
 
-  const toMs = datemath.parse(String(argv.to ?? 'now'))!.valueOf();
-  const to = new Date(toMs);
+  const to = datemath.parse(String(argv.to ?? 'now'))!.valueOf();
 
   const defaultTimeRange = '1m';
 
-  const fromMs = argv.from
+  const from = argv.from
     ? datemath.parse(String(argv.from))!.valueOf()
-    : toMs - intervalToMs(defaultTimeRange);
-  const from = new Date(fromMs);
+    : to - intervalToMs(defaultTimeRange);
 
   const live = argv.live;
 
   if (live) {
-    await startLiveDataUpload({ runOptions, start: from });
+    await startLiveDataUpload({ runOptions, from, to });
   } else {
     await startHistoricalDataUpload({ runOptions, from, to });
   }