Change _source to fields

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/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/get_error_groups/get_error_group_sample_ids.ts

@@ -75,8 +75,7 @@ export async function getErrorGroupSampleIds({
     },
   });
   const errorSampleIds = resp.hits.hits.map((item) => {
-    const source = item.fields;
-    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

@@ -61,15 +61,14 @@ export async function getErrorSampleDetails({
           ],
         },
       },
+      fields: ['*'],
     },
-    fields: ['*'],
   };
 
   const resp = await apmEventClient.search('get_error_sample_details', params);
-  // const error = resp.hits.hits[0]?._source;
 
-  const error = resp.hits.hits[0]?.fields;
-  const errorNorm = normalizeFields(error);
+  // 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;
 

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,

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

@@ -114,6 +114,7 @@ export async function getServiceMetadataDetails({
       track_total_hits: 1,
       size: 1,
       _source: [SERVICE, AGENT, HOST, CONTAINER, KUBERNETES, CLOUD, LABEL_TELEMETRY_AUTO_VERSION],
+      fields: [SERVICE, AGENT, HOST, CONTAINER, KUBERNETES, CLOUD, LABEL_TELEMETRY_AUTO_VERSION],
       query: { bool: { filter, should } },
       aggs: {
         serviceVersions: {

x-pack/plugins/observability_solution/apm/server/routes/span_links/get_linked_parents.ts

@@ -39,6 +39,7 @@ export async function getLinkedParentsOfSpan({
     },
     _source: [SPAN_LINKS],
     body: {
+      fields: [SPAN_LINKS],
       track_total_hits: false,
       size: 1,
       query: {

x-pack/plugins/observability_solution/apm/server/routes/span_links/get_span_links_details.ts

@@ -29,6 +29,7 @@ import { SpanRaw } from '../../../typings/es_schemas/raw/span_raw';
 import { TransactionRaw } from '../../../typings/es_schemas/raw/transaction_raw';
 import { getBufferedTimerange } from './utils';
 import { APMEventClient } from '../../lib/helpers/create_es_client/create_apm_event_client';
+import { normalizeFields } from '../../utils/normalize_fields';
 
 async function fetchSpanLinksDetails({
   apmEventClient,
@@ -68,6 +69,21 @@ async function fetchSpanLinksDetails({
       SERVICE_ENVIRONMENT,
     ],
     body: {
+      fields: [
+        TRACE_ID,
+        SPAN_ID,
+        TRANSACTION_ID,
+        SERVICE_NAME,
+        SPAN_NAME,
+        TRANSACTION_NAME,
+        TRANSACTION_DURATION,
+        SPAN_DURATION,
+        PROCESSOR_EVENT,
+        SPAN_SUBTYPE,
+        SPAN_TYPE,
+        AGENT_NAME,
+        SERVICE_ENVIRONMENT,
+      ],
       track_total_hits: false,
       size: 1000,
       query: {
@@ -106,11 +122,11 @@ async function fetchSpanLinksDetails({
 
   const spanIdsMap = keyBy(spanLinks, 'span.id');
 
-  return response.hits.hits.filter(({ _source: source }) => {
+  return response.hits.hits.filter(({ fields }) => {
     // The above query might return other spans from the same transaction because siblings spans share the same transaction.id
     // so, if it is a span we need to guarantee that the span.id is the same as the span links ids
-    if (source.processor.event === ProcessorEvent.span) {
-      const span = source as SpanRaw;
+    if (fields['processor.event']?.[0] === ProcessorEvent.span) {
+      const span = normalizeFields(fields) as unknown as SpanRaw;
       const hasSpanId = spanIdsMap[span.span.id] || false;
       return hasSpanId;
     }

x-pack/plugins/observability_solution/apm/server/routes/traces/get_trace_items.ts

@@ -114,18 +114,18 @@ export async function getTraceItems({
           must_not: { terms: { [ERROR_LOG_LEVEL]: excludedLogLevels } },
         },
       },
+      fields: [
+        TIMESTAMP,
+        TRACE_ID,
+        TRANSACTION_ID,
+        PARENT_ID,
+        SERVICE_NAME,
+        ERROR_ID,
+        ERROR_LOG_MESSAGE,
+        ERROR_EXCEPTION,
+        ERROR_GROUP_ID,
+      ],
     },
-    fields: [
-      TIMESTAMP,
-      TRACE_ID,
-      TRANSACTION_ID,
-      PARENT_ID,
-      SERVICE_NAME,
-      ERROR_ID,
-      ERROR_LOG_MESSAGE,
-      ERROR_EXCEPTION,
-      ERROR_GROUP_ID,
-    ],
   });
 
   const traceResponsePromise = getTraceDocsPaginated({
@@ -146,8 +146,12 @@ export async function getTraceItems({
   const traceDocsTotal = traceResponse.total;
   const exceedsMax = traceDocsTotal > maxTraceItems;
 
-  const traceDocs = traceResponse.hits.map((hit) => normalizeFields(hit.fields));
-  const errorDocs = errorResponse.hits.hits.map((hit) => normalizeFields(hit.fields));
+  const traceDocs = traceResponse.hits.map(
+    (hit) => normalizeFields(hit.fields) as unknown as WaterfallTransaction | WaterfallSpan
+  );
+  const errorDocs = errorResponse.hits.hits.map(
+    (hit) => normalizeFields(hit.fields) as unknown as WaterfallError
+  );
 
   return {
     exceedsMax,

x-pack/plugins/observability_solution/apm/server/utils/normalize_fields.ts

@@ -14,8 +14,13 @@ export const normalizeFields = (
 
   for (const [key, value] of Object.entries(fields)) {
     const normalizedValue = Array.isArray(value) && value.length > 0 ? value[0] : value;
+    // This function will be fixed in https://github.com/elastic/kibana/issues/192749
+    const arrayValue =
+      Array.isArray(value) && (value.length > 1 || key === 'process.args')
+        ? value
+        : normalizedValue;
 
-    set(normalizedFields, key, normalizedValue);
+    set(normalizedFields, key, arrayValue);
   }
 
   return normalizedFields;