Fix service summary overflow bucket regression

Fix regression introduced in elastic#10061 where service summary overflow
bucket did not work.

Author: carsonip

apmpackage/apm/data_stream/service_summary_interval_metrics/elasticsearch/ingest_pipeline/default.yml

@@ -7,3 +7,6 @@ processors:
       name: observer_ids
   - pipeline:
       name: ecs_version
+  - remove:
+      field: _metric_descriptions
+      ignore_missing: true

systemtest/aggregation_test.go

@@ -328,6 +328,51 @@ func TestServiceSummaryMetricsAggregation(t *testing.T) {
 	systemtest.ApproveEvents(t, t.Name(), result.Hits.Hits)
 }
 
+func TestServiceSummaryMetricsAggregationOverflow(t *testing.T) {
+	systemtest.CleanupElasticsearch(t)
+	srv := apmservertest.NewUnstartedServerTB(t)
+	srv.Config.Aggregation = &apmservertest.AggregationConfig{
+		ServiceTransactionMaxGroups: 2,
+	}
+	require.NoError(t, srv.Start())
+
+	f := func(env string) {
+		t.Setenv("ELASTIC_APM_ENVIRONMENT", env)
+		timestamp, _ := time.Parse(time.RFC3339Nano, "2006-01-02T15:04:05.999999999Z") // should be truncated to 1s
+		tracer := srv.Tracer()
+		tx := tracer.StartTransactionOptions("name", "type", apm.TransactionOptions{Start: timestamp})
+		tx.Duration = time.Second
+		tx.End()
+		tracer.Flush(nil)
+	}
+	f("dev")
+	f("staging")
+	f("prod")
+	f("test")
+
+	// Wait for the transaction to be indexed, indicating that Elasticsearch
+	// indices have been setup and we should not risk triggering the shutdown
+	// timeout while waiting for the aggregated metrics to be indexed.
+	systemtest.Elasticsearch.ExpectMinDocs(t, 4, "traces-apm*",
+		estest.TermQuery{Field: "processor.event", Value: "transaction"},
+	)
+	// Stop server to ensure metrics are flushed on shutdown.
+	assert.NoError(t, srv.Close())
+	systemtest.Elasticsearch.ExpectMinDocs(t, 3, "metrics-apm.service_summary*",
+		estest.BoolQuery{
+			Must: []interface{}{
+				estest.TermQuery{Field: "metricset.name", Value: "service_summary"},
+				estest.TermQuery{Field: "service.name", Value: "_other"},
+			},
+		},
+	)
+	result := systemtest.Elasticsearch.ExpectMinDocs(t, 9, "metrics-apm.service_summary*",
+		estest.TermQuery{Field: "metricset.name", Value: "service_summary"},
+	)
+	// Ignore timestamp because overflow bucket uses time.Now()
+	systemtest.ApproveEvents(t, t.Name(), result.Hits.Hits, "@timestamp")
+}
+
 func TestNonDefaultRollupIntervalHiddenDataStream(t *testing.T) {
 	systemtest.CleanupElasticsearch(t)
 	srv := apmservertest.NewServerTB(t)

systemtest/apmservertest/config.go

@@ -72,6 +72,9 @@ type Config struct {
 
 	// ProfilingConfig holds configuration related to profiling.
 	Profiling *ProfilingConfig `json:"apm-server.profiling,omitempty"`
+
+	// AggregationConfig holds configuration related to aggregation.
+	Aggregation *AggregationConfig `json:"apm-server.aggregation,omitempty"`
 }
 
 // Args formats cfg as a list of arguments to pass to apm-server,
@@ -320,6 +323,11 @@ func durationString(d time.Duration) string {
 	return d.String()
 }
 
+// AggregationConfig holds configuration related to aggregation.
+type AggregationConfig struct {
+	ServiceTransactionMaxGroups int `json:"service_transactions.max_groups,omitempty"`
+}
+
 func configArgs(cfg Config, extra map[string]interface{}) ([]string, error) {
 	kv, err := flattenConfig(cfg, extra)
 	if err != nil {