[APM]Handle ELASTIC_PROFILER_STACK_TRACE_IDS for apm-profiler integration (#217020)

Depends on elastic/elasticsearch#125608

# Summary

`ELASTIC_PROFILER_STACK_TRACE_IDS` is introduced for OTel based data
streams. The same information is stored in
`TRANSACTION_PROFILER_STACK_TRACE_IDS` in the classic APM data streams.

Prior to this PR apm<->profiling integration did not work for OTel SDKs.
This PR adds handling for the new field name.

<img width="1159" alt="Screenshot 2025-04-03 at 10 05 28"
src="https://github.com/user-attachments/assets/ce3ad092-d4f4-4a16-843e-923c72938fe1"
/>

<img width="1772" alt="Screenshot 2025-04-03 at 10 05 40"
src="https://github.com/user-attachments/assets/8b2682fe-6f2e-49a4-9995-d83997a05f02"
/>

---------

Co-authored-by: Greg Kalapos <[email protected]>

Author: cauemarcondes

x-pack/platform/packages/shared/kbn-apm-types/src/es_fields/apm.ts

@@ -74,6 +74,9 @@ export const TRANSACTION_OVERFLOW_COUNT = 'transaction.aggregation.overflow_coun
 export const TRANSACTION_ROOT = 'transaction.root';
 export const TRANSACTION_PROFILER_STACK_TRACE_IDS = 'transaction.profiler_stack_trace_ids';
 
+// OTel field to link profiling and APM
+export const ELASTIC_PROFILER_STACK_TRACE_IDS = 'elastic.profiler_stack_trace_ids';
+
 export const EVENT_OUTCOME = 'event.outcome';
 
 export const TRACE_ID = 'trace.id';

x-pack/solutions/observability/plugins/apm/common/es_fields/__snapshots__/es_fields.test.ts.snap

@@ -0,0 +1,76 @@
+/*
+ * 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; you may not use this file except in compliance with the Elastic License
+ * 2.0.
+ */
+import { ProcessorEvent } from '@kbn/observability-plugin/common';
+import { rangeQuery, termQuery, kqlQuery } from '@kbn/observability-plugin/server';
+import { isEmpty } from 'lodash';
+import { unflattenKnownApmEventFields } from '@kbn/apm-data-access-plugin/server/utils';
+import {
+  ELASTIC_PROFILER_STACK_TRACE_IDS,
+  SERVICE_NAME,
+  TRANSACTION_NAME,
+  TRANSACTION_PROFILER_STACK_TRACE_IDS,
+  TRANSACTION_TYPE,
+} from '../../../common/es_fields/apm';
+import { environmentQuery } from '../../../common/utils/environment_query';
+import type { APMEventClient } from '../../lib/helpers/create_es_client/create_apm_event_client';
+
+export async function getStacktracesIdsField({
+  apmEventClient,
+  start,
+  end,
+  environment,
+  serviceName,
+  transactionType,
+  transactionName,
+  kuery,
+}: {
+  apmEventClient: APMEventClient;
+  start: number;
+  end: number;
+  environment: string;
+  serviceName: string;
+  transactionType: string;
+  transactionName?: string;
+  kuery?: string;
+}) {
+  const response = await apmEventClient.search('get_stacktraces_ids_field', {
+    apm: {
+      events: [ProcessorEvent.transaction],
+    },
+    size: 1,
+    terminate_after: 1,
+    track_total_hits: false,
+    fields: [ELASTIC_PROFILER_STACK_TRACE_IDS, TRANSACTION_PROFILER_STACK_TRACE_IDS],
+    _source: false,
+    query: {
+      bool: {
+        filter: [
+          ...rangeQuery(start, end),
+          ...termQuery(SERVICE_NAME, serviceName),
+          ...termQuery(TRANSACTION_TYPE, transactionType),
+          ...termQuery(TRANSACTION_NAME, transactionName),
+          ...kqlQuery(kuery),
+          ...environmentQuery(environment),
+        ],
+        should: [
+          { exists: { field: ELASTIC_PROFILER_STACK_TRACE_IDS } },
+          { exists: { field: TRANSACTION_PROFILER_STACK_TRACE_IDS } },
+        ],
+      },
+    },
+  });
+
+  const field = unflattenKnownApmEventFields(response.hits.hits[0]?.fields, [
+    ELASTIC_PROFILER_STACK_TRACE_IDS,
+  ]);
+
+  if (!isEmpty(field.elastic.profiler_stack_trace_ids)) {
+    return ELASTIC_PROFILER_STACK_TRACE_IDS;
+  }
+
+  return TRANSACTION_PROFILER_STACK_TRACE_IDS;
+}

x-pack/solutions/observability/plugins/apm/server/routes/profiling/get_stacktraces_ids_field.ts

@@ -5,16 +5,16 @@
  * 2.0.
  */
 
-import { isoToEpochSecsRt, toNumberRt } from '@kbn/io-ts-utils';
+import { toNumberRt } from '@kbn/io-ts-utils';
+import { ProcessorEvent } from '@kbn/observability-plugin/common';
 import type { BaseFlameGraph, TopNFunctions } from '@kbn/profiling-utils';
 import * as t from 'io-ts';
-import { ProcessorEvent } from '@kbn/observability-plugin/common';
 import { getApmEventClient } from '../../lib/helpers/get_apm_event_client';
 import { createApmServerRoute } from '../apm_routes/create_apm_server_route';
-import { environmentRt, kueryRt } from '../default_api_types';
+import { environmentRt, kueryRt, rangeRt } from '../default_api_types';
 import { fetchFlamegraph } from './fetch_flamegraph';
 import { fetchFunctions } from './fetch_functions';
-import { TRANSACTION_PROFILER_STACK_TRACE_IDS } from '../../../common/es_fields/apm';
+import { getStacktracesIdsField } from './get_stacktraces_ids_field';
 
 const servicesFlamegraphRoute = createApmServerRoute({
   endpoint: 'GET /internal/apm/services/{serviceName}/profiling/flamegraph',
@@ -23,12 +23,11 @@ const servicesFlamegraphRoute = createApmServerRoute({
     query: t.intersection([
       kueryRt,
       environmentRt,
+      rangeRt,
       t.partial({
         transactionName: t.string,
       }),
       t.type({
-        start: isoToEpochSecsRt,
-        end: isoToEpochSecsRt,
         transactionType: t.string,
       }),
     ]),
@@ -47,20 +46,30 @@ const servicesFlamegraphRoute = createApmServerRoute({
       const { start, end, kuery, transactionName, transactionType, environment } = params.query;
 
       const indices = apmEventClient.getIndicesFromProcessorEvent(ProcessorEvent.transaction);
+      const stacktraceIdsField = await getStacktracesIdsField({
+        apmEventClient,
+        start,
+        end,
+        environment,
+        serviceName,
+        transactionType,
+        transactionName,
+        kuery,
+      });
 
       return fetchFlamegraph({
         profilingDataAccessStart,
         core,
         esClient: esClient.asCurrentUser,
-        start,
-        end,
+        start: start / 1000,
+        end: end / 1000,
         kuery,
         serviceName,
         transactionName,
         environment,
         transactionType,
         indices,
-        stacktraceIdsField: TRANSACTION_PROFILER_STACK_TRACE_IDS,
+        stacktraceIdsField,
       });
     }
 
@@ -74,12 +83,11 @@ const servicesFunctionsRoute = createApmServerRoute({
     path: t.type({ serviceName: t.string }),
     query: t.intersection([
       environmentRt,
+      rangeRt,
       t.partial({
         transactionName: t.string,
       }),
       t.type({
-        start: isoToEpochSecsRt,
-        end: isoToEpochSecsRt,
         startIndex: toNumberRt,
         endIndex: toNumberRt,
         transactionType: t.string,
@@ -111,6 +119,16 @@ const servicesFunctionsRoute = createApmServerRoute({
       const { serviceName } = params.path;
 
       const indices = apmEventClient.getIndicesFromProcessorEvent(ProcessorEvent.transaction);
+      const stacktraceIdsField = await getStacktracesIdsField({
+        apmEventClient,
+        start,
+        end,
+        environment,
+        serviceName,
+        transactionType,
+        transactionName,
+        kuery,
+      });
 
       return fetchFunctions({
         profilingDataAccessStart,
@@ -119,9 +137,9 @@ const servicesFunctionsRoute = createApmServerRoute({
         startIndex,
         endIndex,
         indices,
-        stacktraceIdsField: TRANSACTION_PROFILER_STACK_TRACE_IDS,
-        start,
-        end,
+        stacktraceIdsField,
+        start: start / 1000,
+        end: end / 1000,
         kuery,
         serviceName,
         transactionName,