aggregation/spanmetrics: handle composite spans

Update spanmetrics to take into account composite spans,
multiplying the span count by the composite count and
using the reported composite span duration instead of
the "gross" duration.

x-pack/apm-server/aggregation/spanmetrics/aggregator.go

@@ -198,16 +198,26 @@ func (a *Aggregator) processSpan(event *model.APMEvent) model.APMEvent {
 		return model.APMEvent{}
 	}
 
+	// For composite spans we use the composite sum duration, which is the sum of
+	// pre-aggregated spans and excludes time gaps that are counted in the reported
+	// span duration. For non-composite spans we just use the reported span duration.
+	count := 1
+	durationMillis := event.Span.Duration
+	if event.Span.Composite != nil {
+		count = event.Span.Composite.Count
+		durationMillis = event.Span.Composite.Sum
+	}
+	duration := time.Duration(durationMillis * float64(time.Millisecond))
+
 	key := aggregationKey{
 		serviceEnvironment: event.Service.Environment,
 		serviceName:        event.Service.Name,
 		agentName:          event.Agent.Name,
 		outcome:            event.Event.Outcome,
 		resource:           event.Span.DestinationService.Resource,
 	}
-	duration := time.Duration(event.Span.Duration * float64(time.Millisecond))
 	metrics := spanMetrics{
-		count: event.Span.RepresentativeCount,
+		count: float64(count) * event.Span.RepresentativeCount,
 		sum:   float64(duration.Microseconds()) * event.Span.RepresentativeCount,
 	}
 	if a.active.storeOrUpdate(key, metrics) {

x-pack/apm-server/aggregation/spanmetrics/aggregator_test.go

@@ -183,6 +183,45 @@ func TestAggregatorRun(t *testing.T) {
 	}
 }
 
+func TestAggregateCompositeSpan(t *testing.T) {
+	batches := make(chan model.Batch, 1)
+	agg, err := NewAggregator(AggregatorConfig{
+		BatchProcessor: makeChanBatchProcessor(batches),
+		Interval:       10 * time.Millisecond,
+		MaxGroups:      1000,
+	})
+	require.NoError(t, err)
+
+	span := makeSpan("service-A", "java", "final_destination", "success", time.Second, 2)
+	span.Span.Composite = &model.Composite{Count: 25, Sum: 700 /* milliseconds */}
+	err = agg.ProcessBatch(context.Background(), &model.Batch{span})
+	require.NoError(t, err)
+
+	// Start the aggregator after processing to ensure metrics are aggregated deterministically.
+	go agg.Run()
+	defer agg.Stop(context.Background())
+
+	batch := expectBatch(t, batches)
+	metricsets := batchMetricsets(t, batch)
+
+	assert.Equal(t, []model.APMEvent{{
+		Agent:   model.Agent{Name: "java"},
+		Service: model.Service{Name: "service-A"},
+		Event:   model.Event{Outcome: "success"},
+		Metricset: &model.Metricset{
+			Name: "service_destination",
+			Span: model.MetricsetSpan{
+				DestinationService: model.DestinationService{Resource: "final_destination"},
+			},
+			Samples: map[string]model.MetricsetSample{
+				"span.destination.service.response_time.count":  {Value: 50.0},
+				"span.destination.service.response_time.sum.us": {Value: 1400000},
+				"metricset.period": {Value: 10},
+			},
+		},
+	}}, metricsets)
+}
+
 func TestAggregatorOverflow(t *testing.T) {
 	batches := make(chan model.Batch, 1)
 	agg, err := NewAggregator(AggregatorConfig{