aggregation/txmetrics: include transaction outcome (#4110)

* model: add Metricset.Event

Enable adding "event.outcome" to metric docs.

* aggregation/txmetrics: group by event.outcome

Include Transaction.Outcome in the aggregation key,
and set `event.outcome` in the metrics docs produced.

changelogs/head.asciidoc

@@ -29,3 +29,4 @@ https://github.com/elastic/apm-server/compare/7.9\...master[View commits]
 * Added apm-server.kibana.headers configuration {pull}4087[4087]
 * Add a new Docker image based on UBI minimal 8 to packaging. {pull}4105[4105]
 * Add event.outcome to transactions and spans {pull}4064[4064]
+* Add event.outcome to aggregated transaction metrics {pull}4110[4110]

model/metricset.go

@@ -34,6 +34,7 @@ import (
 const (
 	metricsetProcessorName  = "metric"
 	metricsetDocType        = "metric"
+	metricsetEventKey       = "event"
 	metricsetTransactionKey = "transaction"
 	metricsetSpanKey        = "span"
 )
@@ -53,6 +54,10 @@ type Metricset struct {
 	// the metrics are associated: service, system, etc.
 	Metadata Metadata
 
+	// Event holds information about the event category with which the
+	// metrics are associated.
+	Event MetricsetEventCategorization
+
 	// Transaction holds information about the transaction group with
 	// which the metrics are associated.
 	Transaction MetricsetTransaction
@@ -103,6 +108,16 @@ type Sample struct {
 	Counts []int64
 }
 
+// MetricsetEventCategorization holds ECS Event Categorization fields
+// for inclusion in metrics. Typically these fields will have been
+// included in the metric aggregation logic.
+//
+// See https://www.elastic.co/guide/en/ecs/current/ecs-category-field-values-reference.html
+type MetricsetEventCategorization struct {
+	// Outcome holds the event outcome: "success", "failure", or "unknown".
+	Outcome string
+}
+
 // MetricsetTransaction provides enough information to connect a metricset to the related kind of transactions.
 type MetricsetTransaction struct {
 	// Name holds the transaction name: "GET /foo", etc.
@@ -148,6 +163,9 @@ func (me *Metricset) Transform(ctx context.Context, _ *transform.Config) []beat.
 
 	fields["processor"] = metricsetProcessorEntry
 	me.Metadata.Set(fields)
+	if eventFields := me.Event.fields(); eventFields != nil {
+		utility.DeepUpdate(fields, metricsetEventKey, eventFields)
+	}
 	if transactionFields := me.Transaction.fields(); transactionFields != nil {
 		utility.DeepUpdate(fields, metricsetTransactionKey, transactionFields)
 	}
@@ -168,6 +186,12 @@ func (me *Metricset) Transform(ctx context.Context, _ *transform.Config) []beat.
 	}}
 }
 
+func (e *MetricsetEventCategorization) fields() common.MapStr {
+	var fields mapStr
+	fields.maybeSetString("outcome", e.Outcome)
+	return common.MapStr(fields)
+}
+
 func (t *MetricsetTransaction) fields() common.MapStr {
 	var fields mapStr
 	fields.maybeSetString("type", t.Type)

model/metricset_test.go

@@ -44,6 +44,8 @@ func TestTransform(t *testing.T) {
 
 		spType    = "db"
 		spSubtype = "sql"
+
+		eventOutcome = "success"
 	)
 
 	tests := []struct {
@@ -106,6 +108,28 @@ func TestTransform(t *testing.T) {
 			},
 			Msg: "Payload with extended transaction metadata.",
 		},
+		{
+			Metricset: &Metricset{
+				Metadata:  metadata,
+				Timestamp: timestamp,
+				Samples: []Sample{{
+					Name:  "metric_field",
+					Value: 123,
+				}},
+				Event: MetricsetEventCategorization{
+					Outcome: eventOutcome,
+				},
+			},
+			Output: []common.MapStr{
+				{
+					"processor":    common.MapStr{"event": "metric", "name": "metric"},
+					"service":      common.MapStr{"name": "myservice"},
+					"event":        common.MapStr{"outcome": eventOutcome},
+					"metric_field": 123.0,
+				},
+			},
+			Msg: "Payload with event categorization metadata.",
+		},
 		{
 			Metricset: &Metricset{
 				Metadata:  metadata,

tests/system/rum_transaction_histogram_metrics.approved.json

@@ -8,7 +8,8 @@
             "version": "1.5.0"
         },
         "event": {
-            "ingested": "2020-06-15T03:33:39.684986Z"
+            "ingested": "2020-08-27T06:33:22.311616Z",
+            "outcome": "unknown"
         },
         "observer": {
             "ephemeral_id": "e9e37f05-448f-4afd-8373-af8b6ff2d6cc",
@@ -27,7 +28,7 @@
             "version": "1.0.0"
         },
         "timeseries": {
-            "instance": "apm-agent-js::2ac4d2f17a65a386"
+            "instance": "apm-agent-js::72a14665326532c3"
         },
         "transaction": {
             "duration": {

tests/system/transaction_histogram_metrics.approved.json

@@ -11,7 +11,8 @@
             "version": "1.5.0"
         },
         "event": {
-            "ingested": "2020-04-24T01:53:20.081678Z"
+            "ingested": "2020-08-27T06:33:25.909902Z",
+            "outcome": "unknown"
         },
         "kubernetes": {
             "pod": {
@@ -39,7 +40,7 @@
             "version": "5.1.3"
         },
         "timeseries": {
-            "instance": "1234_service-12a3:GET /api/types:95818d7bc9580d28"
+            "instance": "1234_service-12a3:GET /api/types:c9273c4ef688630"
         },
         "transaction": {
             "duration": {
@@ -70,17 +71,18 @@
             "version": "1.5.0"
         },
         "event": {
-            "ingested": "2020-04-24T01:53:20.083059Z"
+            "ingested": "2020-08-27T06:33:25.909784Z",
+            "outcome": "unknown"
         },
         "kubernetes": {
             "pod": {
                 "name": "pod-name"
             }
         },
         "observer": {
-            "ephemeral_id": "fdbbb18c-f372-4b20-81c1-ab6596a0703a",
-            "hostname": "alloy",
-            "id": "fb7241d5-1c02-43dc-8561-1b8c5f95ad5c",
+            "ephemeral_id": "1ec77a6e-bc75-4b65-a85c-bd4a992ddcf0",
+            "hostname": "goat",
+            "id": "0d4c6f07-5b6c-4e51-8307-b6252c82fea4",
             "type": "apm-server",
             "version": "8.0.0",
             "version_major": 8
@@ -98,7 +100,7 @@
             "version": "5.1.3"
         },
         "timeseries": {
-            "instance": "1234_service-12a3:GET /api/types:2d9c304bd7ae1b2d"
+            "instance": "1234_service-12a3:GET /api/types:69ebb7e1fd1a35ce"
         },
         "transaction": {
             "duration": {
@@ -129,17 +131,18 @@
             "version": "1.5.0"
         },
         "event": {
-            "ingested": "2020-04-24T01:53:20.082443Z"
+            "ingested": "2020-08-27T06:33:25.909664Z",
+            "outcome": "unknown"
         },
         "kubernetes": {
             "pod": {
                 "name": "pod-name"
             }
         },
         "observer": {
-            "ephemeral_id": "fdbbb18c-f372-4b20-81c1-ab6596a0703a",
-            "hostname": "alloy",
-            "id": "fb7241d5-1c02-43dc-8561-1b8c5f95ad5c",
+            "ephemeral_id": "1ec77a6e-bc75-4b65-a85c-bd4a992ddcf0",
+            "hostname": "goat",
+            "id": "0d4c6f07-5b6c-4e51-8307-b6252c82fea4",
             "type": "apm-server",
             "version": "8.0.0",
             "version_major": 8
@@ -157,7 +160,7 @@
             "version": "5.1.3"
         },
         "timeseries": {
-            "instance": "serviceabc:GET /api/types:eaedcae530dae5c2"
+            "instance": "serviceabc:GET /api/types:750fb663ed9307e4"
         },
         "transaction": {
             "duration": {

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

@@ -326,10 +326,11 @@ func (a *Aggregator) makeTransactionAggregationKey(tx *model.Transaction) transa
 	}
 
 	return transactionAggregationKey{
-		traceRoot:         tx.ParentID == "",
-		transactionName:   tx.Name,
-		transactionResult: tx.Result,
-		transactionType:   tx.Type,
+		traceRoot:          tx.ParentID == "",
+		transactionName:    tx.Name,
+		transactionOutcome: tx.Outcome,
+		transactionResult:  tx.Result,
+		transactionType:    tx.Type,
 
 		agentName:          tx.Metadata.Service.Agent.Name,
 		serviceEnvironment: tx.Metadata.Service.Environment,
@@ -367,6 +368,9 @@ func makeMetricset(key transactionAggregationKey, hash uint64, ts time.Time, cou
 			},
 			// TODO(axw) include client.geo.country_iso_code somewhere
 		},
+		Event: model.MetricsetEventCategorization{
+			Outcome: key.transactionOutcome,
+		},
 		Transaction: model.MetricsetTransaction{
 			Name:   key.transactionName,
 			Type:   key.transactionType,
@@ -423,6 +427,7 @@ type transactionAggregationKey struct {
 	serviceName        string
 	serviceVersion     string
 	transactionName    string
+	transactionOutcome string
 	transactionResult  string
 	transactionType    string
 	userAgentName      string
@@ -442,6 +447,7 @@ func (k *transactionAggregationKey) hash() uint64 {
 	h.WriteString(k.serviceName)
 	h.WriteString(k.serviceVersion)
 	h.WriteString(k.transactionName)
+	h.WriteString(k.transactionOutcome)
 	h.WriteString(k.transactionResult)
 	h.WriteString(k.transactionType)
 	h.WriteString(k.userAgentName)

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

@@ -341,6 +341,93 @@ func testHDRHistogramSignificantFigures(t *testing.T, sigfigs int) {
 	})
 }
 
+func TestAggregationFields(t *testing.T) {
+	reqs := make(chan publish.PendingReq, 1)
+	agg, err := txmetrics.NewAggregator(txmetrics.AggregatorConfig{
+		Report:                         makeChanReporter(reqs),
+		MaxTransactionGroups:           1000,
+		MetricsInterval:                100 * time.Millisecond,
+		HDRHistogramSignificantFigures: 1,
+		RUMUserAgentLRUSize:            1,
+	})
+	require.NoError(t, err)
+	go agg.Run()
+	defer agg.Stop(context.Background())
+
+	input := model.Transaction{RepresentativeCount: 1}
+	inputFields := []*string{
+		&input.Name,
+		&input.Outcome,
+		&input.Result,
+		&input.Type,
+		&input.Metadata.Service.Agent.Name,
+		&input.Metadata.Service.Environment,
+		&input.Metadata.Service.Name,
+		&input.Metadata.Service.Version,
+		&input.Metadata.System.Container.ID,
+		&input.Metadata.System.Kubernetes.PodName,
+	}
+
+	var expected []model.Metricset
+	addExpectedCount := func(expectedCount int64) {
+		expected = append(expected, model.Metricset{
+			Metadata: input.Metadata,
+			Event: model.MetricsetEventCategorization{
+				Outcome: input.Outcome,
+			},
+			Transaction: model.MetricsetTransaction{
+				Name:   input.Name,
+				Type:   input.Type,
+				Result: input.Result,
+				Root:   input.ParentID == "",
+			},
+			Samples: []model.Sample{{
+				Name:   "transaction.duration.histogram",
+				Counts: []int64{expectedCount},
+				Values: []float64{0},
+			}},
+		})
+	}
+	for _, field := range inputFields {
+		for _, value := range []string{"something", "anything"} {
+			*field = value
+			assert.Nil(t, agg.AggregateTransaction(&input))
+			assert.Nil(t, agg.AggregateTransaction(&input))
+			addExpectedCount(2)
+		}
+	}
+
+	// Hostname is complex: if any kubernetes fields are set, then
+	// it is taken from Kubernetes.Node.Name, and DetectedHostname
+	// is ignored.
+	input.Metadata.System.Kubernetes.PodName = ""
+	for _, value := range []string{"something", "anything"} {
+		input.Metadata.System.DetectedHostname = value
+		assert.Nil(t, agg.AggregateTransaction(&input))
+		assert.Nil(t, agg.AggregateTransaction(&input))
+		addExpectedCount(2)
+	}
+
+	// ParentID only impacts aggregation as far as grouping root and
+	// non-root traces.
+	for _, value := range []string{"something", "anything"} {
+		input.ParentID = value
+		assert.Nil(t, agg.AggregateTransaction(&input))
+		assert.Nil(t, agg.AggregateTransaction(&input))
+	}
+	addExpectedCount(4)
+
+	var output []model.Metricset
+	req := expectPublish(t, reqs)
+	for _, tf := range req.Transformables {
+		ms := tf.(*model.Metricset)
+		ms.Timestamp = time.Time{}
+		ms.TimeseriesInstanceID = ""
+		output = append(output, *ms)
+	}
+	assert.ElementsMatch(t, expected, output)
+}
+
 func BenchmarkAggregateTransaction(b *testing.B) {
 	agg, err := txmetrics.NewAggregator(txmetrics.AggregatorConfig{
 		Report:                         makeErrReporter(nil),