[APM] Prefer span metrics over span events (elastic#141519)

* [APM] Prefer span metrics over span events

* [APM] Prefer span metrics over span events

* using weighted_avg when span metrics

* fixing span metrics

* throughput fix

* adding api test

(cherry picked from commit dc53a59)

packages/kbn-apm-synthtrace/src/lib/apm/processors/get_span_destination_metrics.ts

@@ -18,6 +18,7 @@ export function getSpanDestinationMetrics(events: ApmFields[]) {
     'service.environment',
     'service.name',
     'span.destination.service.resource',
+    'span.name',
   ]);
 
   return metricsets.map((metricset) => {

x-pack/plugins/apm/public/components/app/dependency_detail_operations/dependency_detail_operations_list/index.tsx

@@ -9,6 +9,7 @@ import { i18n } from '@kbn/i18n';
 import { keyBy } from 'lodash';
 import React from 'react';
 import { useApmPluginContext } from '../../../../context/apm_plugin/use_apm_plugin_context';
+import { useSearchServiceDestinationMetrics } from '../../../../context/time_range_metadata/use_search_service_destination_metrics';
 import { useApmParams } from '../../../../hooks/use_apm_params';
 import { useBreakpoints } from '../../../../hooks/use_breakpoints';
 import { FETCH_STATUS, useFetcher } from '../../../../hooks/use_fetcher';
@@ -66,6 +67,9 @@ export function DependencyDetailOperationsList() {
     urlComparisonEnabled,
   });
 
+  const { searchServiceDestinationMetrics } =
+    useSearchServiceDestinationMetrics({ rangeFrom, rangeTo, kuery });
+
   const primaryStatsFetch = useFetcher(
     (callApmApi) => {
       return callApmApi('GET /internal/apm/dependencies/operations', {
@@ -76,11 +80,19 @@ export function DependencyDetailOperationsList() {
             end,
             environment,
             kuery,
+            searchServiceDestinationMetrics,
           },
         },
       });
     },
-    [dependencyName, start, end, environment, kuery]
+    [
+      dependencyName,
+      start,
+      end,
+      environment,
+      kuery,
+      searchServiceDestinationMetrics,
+    ]
   );
 
   const comparisonStatsFetch = useFetcher(
@@ -99,11 +111,21 @@ export function DependencyDetailOperationsList() {
             offset,
             environment,
             kuery,
+            searchServiceDestinationMetrics,
           },
         },
       });
     },
-    [dependencyName, start, end, offset, environment, kuery, comparisonEnabled]
+    [
+      dependencyName,
+      start,
+      end,
+      offset,
+      environment,
+      kuery,
+      comparisonEnabled,
+      searchServiceDestinationMetrics,
+    ]
   );
 
   const columns: Array<ITableColumn<OperationStatisticsItem>> = [

x-pack/plugins/apm/server/routes/dependencies/get_top_dependency_operations.ts

@@ -10,23 +10,26 @@ import {
   rangeQuery,
   termQuery,
 } from '@kbn/observability-plugin/server';
-import { ProcessorEvent } from '@kbn/observability-plugin/common';
+import { isFiniteNumber } from '@kbn/observability-plugin/common/utils/is_finite_number';
 import {
   EVENT_OUTCOME,
   SPAN_DESTINATION_SERVICE_RESOURCE,
-  SPAN_DURATION,
+  SPAN_DESTINATION_SERVICE_RESPONSE_TIME_COUNT,
+  SPAN_DESTINATION_SERVICE_RESPONSE_TIME_SUM,
   SPAN_NAME,
 } from '../../../common/elasticsearch_fieldnames';
 import { Environment } from '../../../common/environment_rt';
 import { EventOutcome } from '../../../common/event_outcome';
 import { environmentQuery } from '../../../common/utils/environment_query';
 import { getOffsetInMs } from '../../../common/utils/get_offset_in_ms';
-import {
-  calculateThroughputWithInterval,
-  calculateThroughputWithRange,
-} from '../../lib/helpers/calculate_throughput';
-import { getMetricsDateHistogramParams } from '../../lib/helpers/metrics';
+import { calculateThroughputWithRange } from '../../lib/helpers/calculate_throughput';
+import { getBucketSizeForAggregatedTransactions } from '../../lib/helpers/get_bucket_size_for_aggregated_transactions';
 import { Setup } from '../../lib/helpers/setup_request';
+import {
+  getDocumentTypeFilterForServiceDestinationStatistics,
+  getLatencyFieldForServiceDestinationStatistics,
+  getProcessorEventForServiceDestinationStatistics,
+} from '../../lib/helpers/spans/get_is_using_service_destination_metrics';
 import { calculateImpactBuilder } from '../traces/calculate_impact_builder';
 
 const MAX_NUM_OPERATIONS = 500;
@@ -51,6 +54,7 @@ export async function getTopDependencyOperations({
   offset,
   environment,
   kuery,
+  searchServiceDestinationMetrics,
 }: {
   setup: Setup;
   dependencyName: string;
@@ -59,6 +63,7 @@ export async function getTopDependencyOperations({
   offset?: string;
   environment: Environment;
   kuery: string;
+  searchServiceDestinationMetrics: boolean;
 }) {
   const { apmEventClient } = setup;
 
@@ -68,10 +73,25 @@ export async function getTopDependencyOperations({
     offset,
   });
 
+  const { intervalString } = getBucketSizeForAggregatedTransactions({
+    start: startWithOffset,
+    end: endWithOffset,
+    searchAggregatedServiceMetrics: searchServiceDestinationMetrics,
+  });
+
+  const field = getLatencyFieldForServiceDestinationStatistics(
+    searchServiceDestinationMetrics
+  );
+
   const aggs = {
-    duration: {
-      avg: {
-        field: SPAN_DURATION,
+    latency: {
+      ...(searchServiceDestinationMetrics
+        ? { sum: { field: SPAN_DESTINATION_SERVICE_RESPONSE_TIME_SUM } }
+        : { avg: { field } }),
+    },
+    count: {
+      sum: {
+        field: SPAN_DESTINATION_SERVICE_RESPONSE_TIME_COUNT,
       },
     },
     successful: {
@@ -94,7 +114,11 @@ export async function getTopDependencyOperations({
     'get_top_dependency_operations',
     {
       apm: {
-        events: [ProcessorEvent.span],
+        events: [
+          getProcessorEventForServiceDestinationStatistics(
+            searchServiceDestinationMetrics
+          ),
+        ],
       },
       body: {
         track_total_hits: false,
@@ -106,6 +130,9 @@ export async function getTopDependencyOperations({
               ...environmentQuery(environment),
               ...kqlQuery(kuery),
               ...termQuery(SPAN_DESTINATION_SERVICE_RESOURCE, dependencyName),
+              ...getDocumentTypeFilterForServiceDestinationStatistics(
+                searchServiceDestinationMetrics
+              ),
             ],
           },
         },
@@ -117,18 +144,20 @@ export async function getTopDependencyOperations({
             },
             aggs: {
               over_time: {
-                date_histogram: getMetricsDateHistogramParams({
-                  start: startWithOffset,
-                  end: endWithOffset,
-                  metricsInterval: 60,
-                }),
+                date_histogram: {
+                  field: '@timestamp',
+                  fixed_interval: intervalString,
+                  min_doc_count: 0,
+                  extended_bounds: {
+                    min: startWithOffset,
+                    max: endWithOffset,
+                  },
+                },
                 aggs,
               },
               ...aggs,
               total_time: {
-                sum: {
-                  field: SPAN_DURATION,
-                },
+                sum: { field },
               },
             },
           },
@@ -154,14 +183,28 @@ export async function getTopDependencyOperations({
 
         bucket.over_time.buckets.forEach((dateBucket) => {
           const x = dateBucket.key + offsetInMs;
+          const latencyValue = isFiniteNumber(dateBucket.latency.value)
+            ? dateBucket.latency.value
+            : 0;
+          const count = isFiniteNumber(dateBucket.count.value)
+            ? dateBucket.count.value
+            : 1;
           timeseries.throughput.push({
             x,
-            y: calculateThroughputWithInterval({
-              value: dateBucket.doc_count,
-              bucketSize: 60,
+            y: calculateThroughputWithRange({
+              start: startWithOffset,
+              end: endWithOffset,
+              value: searchServiceDestinationMetrics
+                ? dateBucket.count.value || 0
+                : dateBucket.doc_count,
             }),
           });
-          timeseries.latency.push({ x, y: dateBucket.duration.value });
+          timeseries.latency.push({
+            x,
+            y: searchServiceDestinationMetrics
+              ? latencyValue / count
+              : dateBucket.latency.value,
+          });
           timeseries.failureRate.push({
             x,
             y:
@@ -174,13 +217,24 @@ export async function getTopDependencyOperations({
           });
         });
 
+        const latencyValue = isFiniteNumber(bucket.latency.value)
+          ? bucket.latency.value
+          : 0;
+        const count = isFiniteNumber(bucket.count.value)
+          ? bucket.count.value
+          : 1;
+
         return {
           spanName: bucket.key as string,
-          latency: bucket.duration.value,
+          latency: searchServiceDestinationMetrics
+            ? latencyValue / count
+            : bucket.latency.value,
           throughput: calculateThroughputWithRange({
             start: startWithOffset,
             end: endWithOffset,
-            value: bucket.doc_count,
+            value: searchServiceDestinationMetrics
+              ? bucket.count.value || 0
+              : bucket.doc_count,
           }),
           failureRate:
             bucket.failure.doc_count > 0 || bucket.successful.doc_count > 0

x-pack/plugins/apm/server/routes/dependencies/route.ts

@@ -492,7 +492,10 @@ const dependencyOperationsRoute = createApmServerRoute({
       environmentRt,
       kueryRt,
       offsetRt,
-      t.type({ dependencyName: t.string }),
+      t.type({
+        dependencyName: t.string,
+        searchServiceDestinationMetrics: toBooleanRt,
+      }),
     ]),
   }),
   handler: async (
@@ -501,7 +504,15 @@ const dependencyOperationsRoute = createApmServerRoute({
     const setup = await setupRequest(resources);
 
     const {
-      query: { dependencyName, start, end, environment, kuery, offset },
+      query: {
+        dependencyName,
+        start,
+        end,
+        environment,
+        kuery,
+        offset,
+        searchServiceDestinationMetrics,
+      },
     } = resources.params;
 
     const operations = await getTopDependencyOperations({
@@ -512,6 +523,7 @@ const dependencyOperationsRoute = createApmServerRoute({
       offset,
       environment,
       kuery,
+      searchServiceDestinationMetrics,
     });
 
     return { operations };

x-pack/test/apm_api_integration/tests/dependencies/top_operations.spec.ts

@@ -8,6 +8,8 @@ import expect from '@kbn/expect';
 import { APIReturnType } from '@kbn/apm-plugin/public/services/rest/create_call_apm_api';
 import { ENVIRONMENT_ALL } from '@kbn/apm-plugin/common/environment_filter_values';
 import { ValuesType } from 'utility-types';
+import { DependencyOperation } from '@kbn/apm-plugin/server/routes/dependencies/get_top_dependency_operations';
+import { meanBy } from 'lodash';
 import { FtrProviderContext } from '../../common/ftr_provider_context';
 import { roundNumber } from '../../utils';
 import { generateOperationData, generateOperationDataConfig } from './generate_operation_data';
@@ -37,10 +39,12 @@ export default function ApiTest({ getService }: FtrProviderContext) {
     dependencyName,
     environment = ENVIRONMENT_ALL.value,
     kuery = '',
+    searchServiceDestinationMetrics = false,
   }: {
     dependencyName: string;
     environment?: string;
     kuery?: string;
+    searchServiceDestinationMetrics?: boolean;
   }) {
     return await apmApiClient
       .readUser({
@@ -52,6 +56,7 @@ export default function ApiTest({ getService }: FtrProviderContext) {
             environment,
             kuery,
             dependencyName,
+            searchServiceDestinationMetrics,
           },
         },
       })
@@ -210,5 +215,60 @@ export default function ApiTest({ getService }: FtrProviderContext) {
         expect(bulkOperation).to.be.ok();
       });
     });
+
+    describe('Compare span metrics and span events', () => {
+      let bulkOperationSpanEventsResponse: ValuesType<TopOperations>;
+      let bulkOperationSpanMetricsResponse: ValuesType<TopOperations>;
+
+      before(async () => {
+        const [spanEventsResponse, spanMetricsResponse] = await Promise.all([
+          callApi({ dependencyName: 'elasticsearch', searchServiceDestinationMetrics: false }),
+          callApi({ dependencyName: 'elasticsearch', searchServiceDestinationMetrics: true }),
+        ]);
+        function findBulkOperation(op: DependencyOperation) {
+          return op.spanName === '/_bulk';
+        }
+        bulkOperationSpanEventsResponse = spanEventsResponse.find(findBulkOperation)!;
+        bulkOperationSpanMetricsResponse = spanMetricsResponse.find(findBulkOperation)!;
+      });
+
+      it('returns same latency', () => {
+        expect(bulkOperationSpanEventsResponse.latency).to.eql(
+          bulkOperationSpanMetricsResponse.latency
+        );
+
+        const meanSpanMetrics = meanBy(
+          bulkOperationSpanEventsResponse.timeseries.latency.filter(({ y }) => y !== null),
+          'y'
+        );
+        const meanSpanEvents = meanBy(
+          bulkOperationSpanMetricsResponse.timeseries.latency.filter(({ y }) => y !== null),
+          'y'
+        );
+        expect(meanSpanMetrics).to.eql(meanSpanEvents);
+      });
+
+      it('returns same throughput', () => {
+        expect(bulkOperationSpanEventsResponse.throughput).to.eql(
+          bulkOperationSpanMetricsResponse.throughput
+        );
+
+        const meanSpanMetrics = meanBy(
+          bulkOperationSpanEventsResponse.timeseries.throughput.filter(({ y }) => y !== 0),
+          'y'
+        );
+        const meanSpanEvents = meanBy(
+          bulkOperationSpanMetricsResponse.timeseries.throughput.filter(({ y }) => y !== 0),
+          'y'
+        );
+        expect(meanSpanMetrics).to.eql(meanSpanEvents);
+      });
+
+      it('returns same impact', () => {
+        expect(bulkOperationSpanEventsResponse.impact).to.eql(
+          bulkOperationSpanMetricsResponse.impact
+        );
+      });
+    });
   });
 }