🚧 WIP 🚧 Replace _source in the APM queries

x-pack/plugins/observability_solution/apm/common/correlations/types.ts

@@ -28,6 +28,7 @@ export interface ResponseHitSource {
 
 export interface ResponseHit {
   _source: ResponseHitSource;
+  fields: ResponseHitSource;
 }
 
 export interface CommonCorrelationsQueryParams {

x-pack/plugins/observability_solution/apm/public/components/app/transaction_details/waterfall_with_summary/waterfall_container/waterfall/waterfall_helpers/waterfall_helpers.ts

@@ -174,7 +174,8 @@ function getErrorItem(
   const errorItem: IWaterfallError = {
     docType: 'error',
     doc: error,
-    id: error.error.id,
+    // TODO FIX ME
+    id: error?.error?.id ?? '',
     parent,
     parentId: parent?.id,
     offset: error.timestamp.us - entryTimestamp,

x-pack/plugins/observability_solution/apm/server/lib/apm_telemetry/collect_data_telemetry/tasks.ts

@@ -11,6 +11,7 @@ import { createHash } from 'crypto';
 import { flatten, merge, pickBy, sortBy, sum, uniq } from 'lodash';
 import { SavedObjectsClient } from '@kbn/core/server';
 import type { APMIndices } from '@kbn/apm-data-access-plugin/server';
+import { normalizeFields } from '../../../utils/normalize_fields';
 import { AGENT_NAMES, RUM_AGENT_NAMES } from '../../../../common/agent_name';
 import {
   AGENT_ACTIVATION_METHOD,
@@ -690,10 +691,14 @@ export const tasks: TelemetryTask[] = [
           sort: {
             '@timestamp': 'desc',
           },
+          fields: ['*'],
         },
       });
 
-      const hit = response.hits.hits[0]?._source as Pick<Transaction | Span | APMError, 'observer'>;
+      const hit = normalizeFields(response.hits.hits[0]?.fields) as Pick<
+        Transaction | Span | APMError,
+        'observer'
+      >;
 
       if (!hit || !hit.observer?.version) {
         return {};

x-pack/plugins/observability_solution/apm/server/routes/assistant_functions/get_observability_alert_details_context/get_downstream_dependency_name.ts

@@ -50,9 +50,10 @@ export async function getDownstreamServiceResource({
           ],
         },
       },
+      fields: ['span.destination.service'],
     },
   });
 
   const hit = response.hits.hits[0];
-  return hit?._source?.span.destination?.service.resource;
+  return hit?.fields?.['span.destination?.service?.resource']?.[0] as string;
 }

x-pack/plugins/observability_solution/apm/server/routes/dependencies/get_metadata_for_dependency.ts

@@ -7,9 +7,11 @@
 
 import { rangeQuery } from '@kbn/observability-plugin/server';
 import { ProcessorEvent } from '@kbn/observability-plugin/common';
+import type { Span } from '@kbn/apm-types/es_schemas_ui';
 import { maybe } from '../../../common/utils/maybe';
 import { SPAN_DESTINATION_SERVICE_RESOURCE } from '../../../common/es_fields/apm';
 import { APMEventClient } from '../../lib/helpers/create_es_client/create_apm_event_client';
+import { normalizeFields } from '../../utils/normalize_fields';
 
 export interface MetadataForDependencyResponse {
   spanType: string | undefined;
@@ -46,16 +48,18 @@ export async function getMetadataForDependency({
           ],
         },
       },
+      fields: ['*'],
       sort: {
         '@timestamp': 'desc',
       },
     },
   });
 
-  const sample = maybe(sampleResponse.hits.hits[0])?._source;
+  const sample = maybe(sampleResponse.hits.hits[0])?.fields;
+  const sampleNorm = sample ? (normalizeFields(sample) as unknown as Span) : null;
 
   return {
-    spanType: sample?.span.type,
-    spanSubtype: sample?.span.subtype,
+    spanType: sampleNorm?.span.type,
+    spanSubtype: sampleNorm?.span.subtype,
   };
 }

x-pack/plugins/observability_solution/apm/server/routes/dependencies/get_top_dependency_spans.ts

@@ -8,6 +8,8 @@
 import { ProcessorEvent } from '@kbn/observability-plugin/common';
 import { kqlQuery, rangeQuery, termQuery, termsQuery } from '@kbn/observability-plugin/server';
 import { keyBy } from 'lodash';
+import { Span } from '@kbn/apm-types/es_schemas_ui';
+import type { TransactionRaw } from '@kbn/apm-types/es_schemas_raw';
 import {
   AGENT_NAME,
   EVENT_OUTCOME,
@@ -28,6 +30,7 @@ import { environmentQuery } from '../../../common/utils/environment_query';
 import { maybe } from '../../../common/utils/maybe';
 import { AgentName } from '../../../typings/es_schemas/ui/fields/agent';
 import { APMEventClient } from '../../lib/helpers/create_es_client/create_apm_event_client';
+import { normalizeFields } from '../../utils/normalize_fields';
 
 const MAX_NUM_SPANS = 1000;
 
@@ -110,9 +113,21 @@ export async function getTopDependencySpans({
           EVENT_OUTCOME,
           '@timestamp',
         ],
+        fields: [
+          SPAN_ID,
+          TRACE_ID,
+          TRANSACTION_ID,
+          SPAN_NAME,
+          SERVICE_NAME,
+          SERVICE_ENVIRONMENT,
+          AGENT_NAME,
+          SPAN_DURATION,
+          EVENT_OUTCOME,
+          '@timestamp',
+        ],
       },
     })
-  ).hits.hits.map((hit) => hit._source);
+  ).hits.hits.map((hit) => normalizeFields(hit?.fields) as unknown as Span);
 
   const transactionIds = spans.map((span) => span.transaction!.id);
 
@@ -130,12 +145,13 @@ export async function getTopDependencySpans({
           },
         },
         _source: [TRANSACTION_ID, TRANSACTION_TYPE, TRANSACTION_NAME],
+        fields: [TRANSACTION_ID, TRANSACTION_TYPE, TRANSACTION_NAME],
         sort: {
           '@timestamp': 'desc',
         },
       },
     })
-  ).hits.hits.map((hit) => hit._source);
+  ).hits.hits.map((hit) => normalizeFields(hit.fields) as unknown as TransactionRaw);
 
   const transactionsById = keyBy(transactions, (transaction) => transaction.transaction.id);
 

x-pack/plugins/observability_solution/apm/server/routes/entities/get_entity_latest_services.ts

@@ -17,6 +17,7 @@ import { environmentQuery } from '../../../common/utils/environment_query';
 import { EntitiesESClient } from '../../lib/helpers/create_es_client/create_entities_es_client/create_entities_es_client';
 import { entitiesRangeQuery } from './get_entities';
 import { EntityLatestServiceRaw, EntityType } from './types';
+import { normalizeFields } from '../../utils/normalize_fields';
 
 export async function getEntityLatestServices({
   entitiesESClient,
@@ -41,6 +42,7 @@ export async function getEntityLatestServices({
         size,
         track_total_hits: false,
         _source: [AGENT_NAME, ENTITY, DATA_STEAM_TYPE, SERVICE_NAME, SERVICE_ENVIRONMENT],
+        fields: [AGENT_NAME, ENTITY, DATA_STEAM_TYPE, SERVICE_NAME, SERVICE_ENVIRONMENT],
         query: {
           bool: {
             filter: [
@@ -54,7 +56,7 @@ export async function getEntityLatestServices({
         },
       },
     })
-  ).hits.hits.map((hit) => hit._source);
+  ).hits.hits.map((hit) => normalizeFields(hit?.fields) as unknown as EntityLatestServiceRaw);
 
   return latestEntityServices;
 }

x-pack/plugins/observability_solution/apm/server/routes/errors/erroneous_transactions/get_top_erroneous_transactions.ts

@@ -94,6 +94,7 @@ async function getTopErroneousTransactions({
               top_hits: {
                 size: 1,
                 _source: [TRANSACTION_TYPE],
+                fields: [TRANSACTION_TYPE],
               },
             },
             timeseries: {
@@ -117,7 +118,7 @@ async function getTopErroneousTransactions({
       ({ key, doc_count: docCount, sample, timeseries }) => {
         return {
           transactionName: key as string,
-          transactionType: sample.hits.hits[0]._source.transaction?.type,
+          transactionType: sample.hits.hits[0].fields?.['transaction.type']?.[0],
           occurrences: docCount,
           timeseries: timeseries.buckets.map((timeseriesBucket) => {
             return {

x-pack/plugins/observability_solution/apm/server/routes/errors/get_error_groups/get_error_group_main_statistics.ts

@@ -7,6 +7,8 @@
 
 import { AggregationsAggregateOrder } from '@elastic/elasticsearch/lib/api/types';
 import { kqlQuery, rangeQuery, termQuery, wildcardQuery } from '@kbn/observability-plugin/server';
+import type { ErrorRaw } from '@kbn/apm-types/es_schemas_raw';
+import { normalizeFields } from '../../../utils/normalize_fields';
 import {
   ERROR_CULPRIT,
   ERROR_EXC_HANDLED,
@@ -139,6 +141,16 @@ export async function getErrorGroupMainStatistics({
                   ERROR_GROUP_ID,
                   '@timestamp',
                 ],
+                fields: [
+                  TRACE_ID,
+                  ERROR_LOG_MESSAGE,
+                  ERROR_EXC_MESSAGE,
+                  ERROR_EXC_HANDLED,
+                  ERROR_EXC_TYPE,
+                  ERROR_CULPRIT,
+                  ERROR_GROUP_ID,
+                  '@timestamp',
+                ],
                 sort: {
                   '@timestamp': 'desc',
                 },
@@ -157,15 +169,17 @@ export async function getErrorGroupMainStatistics({
 
   const errorGroups =
     response.aggregations?.error_groups.buckets.map((bucket) => {
+      const fields = bucket.sample.hits.hits[0]?.fields;
+      const fieldsNorm = normalizeFields(fields) as unknown as ErrorRaw;
       return {
         groupId: bucket.key as string,
-        name: getErrorName(bucket.sample.hits.hits[0]._source),
-        lastSeen: new Date(bucket.sample.hits.hits[0]._source['@timestamp']).getTime(),
+        name: getErrorName(fieldsNorm),
+        lastSeen: new Date(fieldsNorm['@timestamp']).getTime(),
         occurrences: bucket.doc_count,
-        culprit: bucket.sample.hits.hits[0]._source.error.culprit,
-        handled: bucket.sample.hits.hits[0]._source.error.exception?.[0].handled,
-        type: bucket.sample.hits.hits[0]._source.error.exception?.[0].type,
-        traceId: bucket.sample.hits.hits[0]._source.trace?.id,
+        culprit: fieldsNorm.error.culprit,
+        handled: fieldsNorm.error.exception?.[0]?.handled,
+        type: fieldsNorm.error.exception?.[0]?.type,
+        traceId: fieldsNorm.trace?.id,
       };
     }) ?? [];
 

x-pack/plugins/observability_solution/apm/server/routes/errors/get_error_groups/get_error_group_sample_ids.ts

@@ -67,15 +67,15 @@ export async function getErrorGroupSampleIds({
         },
       },
       _source: [ERROR_ID, 'transaction'],
+      fields: [ERROR_ID, 'transaction'],
       sort: asMutableArray([
         { _score: { order: 'desc' } }, // sort by _score first to ensure that errors with transaction.sampled:true ends up on top
         { '@timestamp': { order: 'desc' } }, // sort by timestamp to get the most recent error
       ] as const),
     },
   });
   const errorSampleIds = resp.hits.hits.map((item) => {
-    const source = item._source;
-    return source.error.id;
+    return item.fields['error.id']?.[0] as string;
   });
 
   return {

x-pack/plugins/observability_solution/apm/server/routes/errors/get_error_groups/get_error_sample_details.ts

@@ -6,6 +6,7 @@
  */
 
 import { rangeQuery, kqlQuery } from '@kbn/observability-plugin/server';
+import { normalizeFields } from '../../../utils/normalize_fields';
 import { ERROR_ID, SERVICE_NAME } from '../../../../common/es_fields/apm';
 import { environmentQuery } from '../../../../common/utils/environment_query';
 import { ApmDocumentType } from '../../../../common/document_type';
@@ -60,13 +61,16 @@ export async function getErrorSampleDetails({
           ],
         },
       },
+      fields: ['*'],
     },
   };
 
   const resp = await apmEventClient.search('get_error_sample_details', params);
-  const error = resp.hits.hits[0]?._source;
-  const transactionId = error?.transaction?.id;
-  const traceId = error?.trace?.id;
+
+  // const error = resp.hits.hits[0]?.fields;
+  const errorNorm = normalizeFields(resp.hits.hits[0]?.fields) as unknown as APMError;
+  const transactionId = errorNorm?.transaction?.id;
+  const traceId = errorNorm?.trace?.id;
 
   let transaction;
   if (transactionId && traceId) {
@@ -81,6 +85,6 @@ export async function getErrorSampleDetails({
 
   return {
     transaction,
-    error,
+    error: errorNorm,
   };
 }

x-pack/plugins/observability_solution/apm/server/routes/services/get_service_agent.ts

@@ -16,6 +16,7 @@ import {
 } from '../../../common/es_fields/apm';
 import { APMEventClient } from '../../lib/helpers/create_es_client/create_apm_event_client';
 import { getServerlessTypeFromCloudData, ServerlessType } from '../../../common/serverless';
+import { normalizeFields } from '../../utils/normalize_fields';
 
 interface ServiceAgent {
   agent?: {
@@ -90,6 +91,7 @@ export async function getServiceAgent({
           ],
         },
       },
+      fields: [AGENT_NAME, SERVICE_RUNTIME_NAME, CLOUD_PROVIDER, CLOUD_SERVICE_NAME],
       sort: {
         _score: { order: 'desc' as const },
       },
@@ -101,7 +103,7 @@ export async function getServiceAgent({
     return {};
   }
 
-  const { agent, service, cloud } = response.hits.hits[0]._source as ServiceAgent;
+  const { agent, service, cloud } = normalizeFields(response.hits.hits[0].fields) as ServiceAgent;
   const serverlessType = getServerlessTypeFromCloudData(cloud?.provider, cloud?.service?.name);
 
   return {

x-pack/plugins/observability_solution/apm/server/routes/services/get_service_instance_metadata_details.ts

@@ -20,6 +20,7 @@ import { Container } from '../../../typings/es_schemas/raw/fields/container';
 import { Kubernetes } from '../../../typings/es_schemas/raw/fields/kubernetes';
 import { Host } from '../../../typings/es_schemas/raw/fields/host';
 import { Cloud } from '../../../typings/es_schemas/raw/fields/cloud';
+import { normalizeFields } from '../../utils/normalize_fields';
 
 export interface ServiceInstanceMetadataDetailsResponse {
   '@timestamp': string;
@@ -66,11 +67,16 @@ export async function getServiceInstanceMetadataDetails({
               filter: filter.concat({ term: { [METRICSET_NAME]: 'app' } }),
             },
           },
+          fields: ['*'],
         },
       }
     );
 
-    return maybe(response.hits.hits[0]?._source);
+    return maybe(
+      normalizeFields(
+        response.hits.hits[0]?.fields
+      ) as unknown as ServiceInstanceMetadataDetailsResponse
+    );
   }
 
   async function getTransactionEventSample() {
@@ -85,11 +91,12 @@ export async function getServiceInstanceMetadataDetails({
           terminate_after: 1,
           size: 1,
           query: { bool: { filter } },
+          fields: ['*'],
         },
       }
     );
 
-    return maybe(response.hits.hits[0]?._source);
+    return maybe(normalizeFields(response.hits.hits[0]?.fields));
   }
 
   async function getTransactionMetricSample() {
@@ -108,10 +115,11 @@ export async function getServiceInstanceMetadataDetails({
               filter: filter.concat(getBackwardCompatibleDocumentTypeFilter(true)),
             },
           },
+          fields: ['*'],
         },
       }
     );
-    return maybe(response.hits.hits[0]?._source);
+    return maybe(normalizeFields(response.hits.hits[0]?.fields));
   }
 
   // we can expect the most detail of application metrics,