[Streams] Replay loghub data with synthtrace

.github/CODEOWNERS

@@ -825,6 +825,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/kbn-streams-schema @elastic/streams-program-team
 x-pack/platform/packages/shared/logs-overview @elastic/obs-ux-logs-team

package.json

@@ -1495,6 +1495,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:packages/kbn-scout",
     "@kbn/scout-info": "link:packages/kbn-scout-info",
     "@kbn/scout-oblt": "link:x-pack/solutions/observability/packages/kbn-scout-oblt",

packages/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';

packages/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';

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

@@ -20,6 +20,7 @@ import { Logger } from '../lib/utils/create_logger';
 import { ScenarioReturnType } from '../lib/utils/with_client';
 import { RunOptions } from './utils/parse_run_cli_flags';
 import { EntitiesSynthtraceKibanaClient } from '../lib/entities/entities_synthtrace_kibana_client';
+import { StreamsSynthtraceClient } from '../lib/streams/streams_synthtrace_client';
 
 interface EsClients {
   apmEsClient: ApmSynthtraceEsClient;
@@ -28,6 +29,7 @@ interface EsClients {
   syntheticsEsClient: SyntheticsSynthtraceEsClient;
   otelEsClient: OtelSynthtraceEsClient;
   entitiesEsClient: EntitiesSynthtraceEsClient;
+  streamsClient: StreamsSynthtraceClient;
 }
 
 interface KibanaClients {

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

@@ -7,100 +7,52 @@
  * License v3.0 only", or the "Server Side Public License, v 1".
  */
 
+import { Client } from '@elastic/elasticsearch';
 import { createLogger } from '../../lib/utils/create_logger';
-import { getApmEsClient } from './get_apm_es_client';
-import { getLogsEsClient } from './get_logs_es_client';
-import { getInfraEsClient } from './get_infra_es_client';
+import { getClients } from './get_clients';
 import { getKibanaClient } from './get_kibana_client';
 import { getServiceUrls } from './get_service_urls';
 import { RunOptions } from './parse_run_cli_flags';
-import { getSyntheticsEsClient } from './get_synthetics_es_client';
-import { getOtelSynthtraceEsClient } from './get_otel_es_client';
-import { getEntitiesEsClient } from './get_entities_es_client';
-import { getEntitiesKibanaClient } from './get_entites_kibana_client';
+import { getEsClientTlsSettings } from './ssl';
 
 export async function bootstrap(runOptions: RunOptions) {
   const logger = createLogger(runOptions.logLevel);
 
-  let version = runOptions['assume-package-version'];
-
   const { kibanaUrl, esUrl } = await getServiceUrls({ ...runOptions, logger });
 
   const kibanaClient = getKibanaClient({
     target: kibanaUrl,
     logger,
   });
 
-  if (!version) {
-    version = await kibanaClient.fetchLatestApmPackageVersion();
-    await kibanaClient.installApmPackage(version);
-  } else if (version === 'latest') {
-    version = await kibanaClient.fetchLatestApmPackageVersion();
-  }
-
-  logger.info(`Using package version: ${version}`);
-
-  const apmEsClient = getApmEsClient({
-    target: esUrl,
-    logger,
-    concurrency: runOptions.concurrency,
-    version,
-  });
-
-  const logsEsClient = getLogsEsClient({
-    target: esUrl,
-    logger,
-    concurrency: runOptions.concurrency,
-  });
-
-  const infraEsClient = getInfraEsClient({
-    target: esUrl,
-    logger,
-    concurrency: runOptions.concurrency,
-  });
-
-  const entitiesEsClient = getEntitiesEsClient({
-    target: esUrl,
-    logger,
-    concurrency: runOptions.concurrency,
-  });
-
-  const entitiesKibanaClient = getEntitiesKibanaClient({
-    target: kibanaUrl,
-    logger,
+  const client = new Client({
+    node: esUrl,
+    tls: getEsClientTlsSettings(esUrl),
   });
 
-  const syntheticsEsClient = getSyntheticsEsClient({
-    target: esUrl,
-    logger,
-    concurrency: runOptions.concurrency,
-  });
-  const otelEsClient = getOtelSynthtraceEsClient({
-    target: esUrl,
+  const clients = await getClients({
     logger,
-    concurrency: runOptions.concurrency,
+    packageVersion: runOptions['assume-package-version'],
+    options: {
+      client,
+      logger,
+      concurrency: runOptions.concurrency,
+      kibana: kibanaClient,
+    },
   });
 
   if (runOptions.clean) {
-    await apmEsClient.clean();
-    await logsEsClient.clean();
-    await infraEsClient.clean();
-    await entitiesEsClient.clean();
-    await syntheticsEsClient.clean();
-    await otelEsClient.clean();
+    for (const synthtraceClient of Object.values(clients)) {
+      if ('clean' in synthtraceClient) {
+        await synthtraceClient.clean();
+      }
+    }
   }
 
   return {
+    clients,
     logger,
-    apmEsClient,
-    logsEsClient,
-    infraEsClient,
-    entitiesEsClient,
-    syntheticsEsClient,
-    otelEsClient,
-    version,
     kibanaUrl,
     esUrl,
-    entitiesKibanaClient,
   };
 }

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

@@ -0,0 +1,77 @@
+/*
+ * 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 { Required } from 'utility-types';
+import { ApmSynthtraceEsClient } from '../../lib/apm/client/apm_synthtrace_es_client';
+import { ApmSynthtraceKibanaClient } from '../../lib/apm/client/apm_synthtrace_kibana_client';
+import { EntitiesSynthtraceEsClient } from '../../lib/entities/entities_synthtrace_es_client';
+import { EntitiesSynthtraceKibanaClient } from '../../lib/entities/entities_synthtrace_kibana_client';
+import { InfraSynthtraceEsClient } from '../../lib/infra/infra_synthtrace_es_client';
+import { LogsSynthtraceEsClient } from '../../lib/logs/logs_synthtrace_es_client';
+import { OtelSynthtraceEsClient } from '../../lib/otel/otel_synthtrace_es_client';
+import { SynthtraceEsClientOptions } from '../../lib/shared/base_client';
+import { StreamsSynthtraceClient } from '../../lib/streams/streams_synthtrace_client';
+import { SyntheticsSynthtraceEsClient } from '../../lib/synthetics/synthetics_synthtrace_es_client';
+import { Logger } from '../../lib/utils/create_logger';
+
+export async function getClients({
+  logger,
+  options,
+  packageVersion,
+}: {
+  logger: Logger;
+  options: Required<Omit<SynthtraceEsClientOptions, 'pipeline'>, 'kibana'>;
+  packageVersion?: string;
+}) {
+  const apmKibanaClient = new ApmSynthtraceKibanaClient({
+    logger,
+    kibanaClient: options.kibana,
+  });
+
+  let version = packageVersion;
+
+  if (!version) {
+    version = await apmKibanaClient.fetchLatestApmPackageVersion();
+    await apmKibanaClient.installApmPackage(version);
+  } else if (version === 'latest') {
+    version = await apmKibanaClient.fetchLatestApmPackageVersion();
+  }
+
+  logger.info(`Using package version: ${version}`);
+
+  const apmEsClient = new ApmSynthtraceEsClient({
+    ...options,
+    version,
+  });
+
+  const logsEsClient = new LogsSynthtraceEsClient(options);
+  const infraEsClient = new InfraSynthtraceEsClient(options);
+  const entitiesEsClient = new EntitiesSynthtraceEsClient(options);
+
+  const entitiesKibanaClient = new EntitiesSynthtraceKibanaClient({
+    ...options,
+    kibanaClient: options.kibana,
+  });
+
+  const syntheticsEsClient = new SyntheticsSynthtraceEsClient(options);
+  const otelEsClient = new OtelSynthtraceEsClient(options);
+
+  const streamsClient = new StreamsSynthtraceClient(options);
+
+  return {
+    apmEsClient,
+    entitiesEsClient,
+    infraEsClient,
+    logsEsClient,
+    otelEsClient,
+    streamsClient,
+    syntheticsEsClient,
+    entitiesKibanaClient,
+  };
+}