Add event.outcome_numeric

apmpackage/apm/changelog.yml

@@ -3,6 +3,9 @@
     - description: Enable synthetic source for metrics data streams
       type: enhancement
       link: https://github.com/elastic/apm-server/pull/9756
+    - description: Add `event.outcome_numeric`
+      type: enhancement
+      link: https://github.com/elastic/apm-server/pull/9764
 - version: "8.6.0"
   changes:
     - description: Change `ecs.version` to a `constant_keyword` field

apmpackage/apm/data_stream/internal_metrics/manifest.yml

@@ -25,6 +25,12 @@ elasticsearch:
                   type: aggregate_metric_double
                   metrics: [sum, value_count]
                   default_metric: sum
+        event:
+          properties:
+            outcome_numeric:
+              type: aggregate_metric_double
+              metrics: [ sum, value_count ]
+              default_metric: sum
     settings:
       index:
         sort.field: "@timestamp"

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

@@ -15,6 +15,8 @@ processors:
       name: client_geoip
   - pipeline:
       name: event_duration
+  - pipeline:
+      name: event_outcome_numeric
   - remove:
       # Remove some metadata from spans that is available in the parent transaction, to cut down on storage costs.
       if: ctx.processor?.event == 'span'

apmpackage/apm/data_stream/traces/fields/fields.yml

@@ -18,6 +18,11 @@
   type: keyword
   description: |
     The cloud service name is intended to distinguish services running on different platforms within a provider.
+- name: event.outcome_numeric
+  type: byte
+  index: false
+  description: |
+    Numeric value of event.outcome: 1 for 'success', 0 for 'failure', and null for 'unknown'.
 - name: faas.id
   type: keyword
   description: |

apmpackage/cmd/genpackage/pipelines.go

@@ -29,20 +29,21 @@ import (
 // To use a common pipeline, define a "pipeline" processor with the "name" property set to one of
 // the common pipelines. e.g.
 //
-//   processors:
-//     - ...
-//     - pipeline:
-//         name: observer_version
-//     - ...
+//	processors:
+//	  - ...
+//	  - pipeline:
+//	      name: observer_version
+//	  - ...
 func getCommonPipeline(name string, version *version.V) []map[string]interface{} {
 	commonPipelines := map[string][]map[string]interface{}{
-		"observer_version": getObserverVersionPipeline(version),
-		"observer_ids":     observerIDsPipeline,
-		"ecs_version":      ecsVersionPipeline,
-		"user_agent":       userAgentPipeline,
-		"process_ppid":     processPpidPipeline,
-		"client_geoip":     clientGeoIPPipeline,
-		"event_duration":   eventDurationPipeline,
+		"observer_version":      getObserverVersionPipeline(version),
+		"observer_ids":          observerIDsPipeline,
+		"ecs_version":           ecsVersionPipeline,
+		"user_agent":            userAgentPipeline,
+		"process_ppid":          processPpidPipeline,
+		"client_geoip":          clientGeoIPPipeline,
+		"event_duration":        eventDurationPipeline,
+		"event_outcome_numeric": eventOutcomeNumericPipeline,
 	}
 	return commonPipelines[name]
 }
@@ -167,3 +168,17 @@ ctx.get(ctx.processor.event).duration = ["us": (int)(durationNanos/1000)];
 		"ignore_failure": true,
 	},
 }}
+
+var eventOutcomeNumericPipeline = []map[string]interface{}{{
+	"set": map[string]interface{}{
+		"if":    "ctx.event?.outcome == 'success'",
+		"field": "event.outcome_numeric",
+		"value": 1,
+	},
+}, {
+	"set": map[string]interface{}{
+		"if":    "ctx.event?.outcome == 'failure'",
+		"field": "event.outcome_numeric",
+		"value": 0,
+	},
+}}

internal/model/event.go

@@ -35,6 +35,9 @@ type Event struct {
 	// Outcome holds the event outcome: "success", "failure", or "unknown".
 	Outcome string
 
+	// OutcomeNumeric TODO
+	OutcomeNumeric *SummaryMetric
+
 	// Severity holds the numeric severity of the event for log events.
 	Severity int64
 
@@ -50,6 +53,9 @@ type Event struct {
 func (e *Event) fields() mapstr.M {
 	var fields mapStr
 	fields.maybeSetString("outcome", e.Outcome)
+	if e.OutcomeNumeric != nil {
+		fields.maybeSetMapStr("outcome_numeric", e.OutcomeNumeric.fields())
+	}
 	fields.maybeSetString("action", e.Action)
 	fields.maybeSetString("dataset", e.Dataset)
 	if e.Severity > 0 {

internal/model/modeldecoder/v2/metadata_test.go

@@ -127,6 +127,7 @@ func isUnmappedMetadataField(key string) bool {
 		"Event",
 		"Event.Duration",
 		"Event.Outcome",
+		"Event.OutcomeNumeric",
 		"Event.Dataset",
 		"Event.Severity",
 		"Event.Action",

systemtest/approvals/TestCompressedSpans.approved.json

@@ -12,7 +12,8 @@
             "event": {
                 "agent_id_status": "missing",
                 "ingested": "dynamic",
-                "outcome": "success"
+                "outcome": "success",
+                "outcome_numeric": 1
             },
             "observer": {
                 "hostname": "dynamic",
@@ -76,7 +77,8 @@
             "event": {
                 "agent_id_status": "missing",
                 "ingested": "dynamic",
-                "outcome": "success"
+                "outcome": "success",
+                "outcome_numeric": 1
             },
             "observer": {
                 "hostname": "dynamic",

systemtest/approvals/TestDropUnsampled.approved.json

@@ -12,7 +12,8 @@
             "event": {
                 "agent_id_status": "missing",
                 "ingested": "dynamic",
-                "outcome": "success"
+                "outcome": "success",
+                "outcome_numeric": 1
             },
             "host": {
                 "architecture": "i386",

systemtest/approvals/TestFleetIntegration.approved.json

@@ -12,7 +12,8 @@
             "event": {
                 "agent_id_status": "missing",
                 "ingested": "dynamic",
-                "outcome": "success"
+                "outcome": "success",
+                "outcome_numeric": 1
             },
             "host": {
                 "architecture": "i386",

systemtest/approvals/TestTransactionDroppedSpansStatsTransaction.approved.json

@@ -12,7 +12,8 @@
             "event": {
                 "agent_id_status": "missing",
                 "ingested": "dynamic",
-                "outcome": "success"
+                "outcome": "success",
+                "outcome_numeric": 1
             },
             "host": {
                 "architecture": "i386",

systemtest/ingest_test.go

@@ -226,3 +226,63 @@ func TestECSVersion(t *testing.T) {
 	result := systemtest.Elasticsearch.ExpectDocs(t, "traces-apm*", nil)
 	assert.Equal(t, []interface{}{"8.6.0-dev"}, result.Hits.Hits[0].Fields["ecs.version"])
 }
+
+func TestIngestPipelineEventOutcomeNumeric(t *testing.T) {
+	type test struct {
+		source                   string
+		eventOutcomeNumericNull  bool
+		eventOutcomeNumericValue int
+	}
+
+	tests := []test{
+		{
+			source: `{"@timestamp": "2022-02-15", "observer": {"version": "8.2.0"}, 
+"processor": {"event": "transaction"}}`,
+			eventOutcomeNumericNull: true,
+		},
+		{
+			source: `{"@timestamp": "2022-02-15", "observer": {"version": "8.2.0"}, 
+"processor": {"event": "transaction"}, "event": {"outcome": "success"}}`,
+			eventOutcomeNumericValue: 1,
+		},
+		{
+			source: `{"@timestamp": "2022-02-15", "observer": {"version": "8.2.0"}, 
+"processor": {"event": "transaction"}, "event": {"outcome": "failure"}}`,
+			eventOutcomeNumericValue: 0,
+		},
+		{
+			source: `{"@timestamp": "2022-02-15", "observer": {"version": "8.2.0"}, 
+"processor": {"event": "transaction"}, "event": {"outcome": "unknown"}}`,
+			eventOutcomeNumericNull: true,
+		},
+	}
+
+	for _, test := range tests {
+		var indexResponse struct {
+			Index string `json:"_index"`
+			ID    string `json:"_id"`
+		}
+		_, err := systemtest.Elasticsearch.Do(context.Background(), esapi.IndexRequest{
+			Index:   "traces-apm-default",
+			Body:    strings.NewReader(test.source),
+			Refresh: "true",
+		}, &indexResponse)
+		require.NoError(t, err)
+
+		var doc struct {
+			Source json.RawMessage `json:"_source"`
+		}
+		_, err = systemtest.Elasticsearch.Do(context.Background(), esapi.GetRequest{
+			Index:      indexResponse.Index,
+			DocumentID: indexResponse.ID,
+		}, &doc)
+		require.NoError(t, err)
+
+		outcomeNumeric := gjson.GetBytes(doc.Source, "event.outcome_numeric")
+		if test.eventOutcomeNumericNull {
+			assert.Equal(t, nil, outcomeNumeric.Value())
+		} else {
+			assert.Equal(t, test.eventOutcomeNumericValue, int(outcomeNumeric.Value().(float64)))
+		}
+	}
+}

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

@@ -380,5 +380,11 @@ func makeMetricset(key aggregationKey, metrics serviceMetrics) model.APMEvent {
 				Sum:   float64(time.Duration(math.Round(metrics.transactionDuration)).Microseconds()),
 			},
 		},
+		Event: model.Event{
+			OutcomeNumeric: &model.SummaryMetric{
+				Count: int64(math.Round(metrics.successCount + metrics.failureCount)),
+				Sum:   metrics.successCount,
+			},
+		},
 	}
 }

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

@@ -123,6 +123,12 @@ func TestAggregatorRun(t *testing.T) {
 			SuccessCount: 2,
 			FailureCount: 3,
 		},
+		Event: model.Event{
+			OutcomeNumeric: &model.SummaryMetric{
+				Count: 5,
+				Sum:   2,
+			},
+		},
 	}, {
 		Processor: model.MetricsetProcessor,
 		Metricset: &model.Metricset{Name: "service", DocCount: 1},
@@ -135,6 +141,9 @@ func TestAggregatorRun(t *testing.T) {
 				Sum:   1000, // 1ms in micros
 			},
 		},
+		Event: model.Event{
+			OutcomeNumeric: &model.SummaryMetric{},
+		},
 	}, {
 		Processor: model.MetricsetProcessor,
 		Metricset: &model.Metricset{Name: "service", DocCount: 1},
@@ -147,6 +156,9 @@ func TestAggregatorRun(t *testing.T) {
 				Sum:   1000, // 1ms in micros
 			},
 		},
+		Event: model.Event{
+			OutcomeNumeric: &model.SummaryMetric{},
+		},
 	}, {
 		Processor: model.MetricsetProcessor,
 		Metricset: &model.Metricset{Name: "service", DocCount: 1},
@@ -159,6 +171,9 @@ func TestAggregatorRun(t *testing.T) {
 				Sum:   1000, // 1ms in micros
 			},
 		},
+		Event: model.Event{
+			OutcomeNumeric: &model.SummaryMetric{},
+		},
 	}, {
 		Processor: model.MetricsetProcessor,
 		Metricset: &model.Metricset{Name: "service", DocCount: 1},
@@ -171,6 +186,9 @@ func TestAggregatorRun(t *testing.T) {
 				Sum:   1000, // 1ms in micros
 			},
 		},
+		Event: model.Event{
+			OutcomeNumeric: &model.SummaryMetric{},
+		},
 	}}
 
 	assert.Equal(t, len(expected), len(metricsets))
@@ -282,6 +300,12 @@ func TestAggregatorMaxGroups(t *testing.T) {
 				SuccessCount: 1,
 			},
 			Metricset: &model.Metricset{Name: "service", DocCount: 1},
+			Event: model.Event{
+				OutcomeNumeric: &model.SummaryMetric{
+					Count: 1,
+					Sum:   1,
+				},
+			},
 		}, m)
 	}
 }

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

@@ -390,6 +390,18 @@ func (a *Aggregator) makeTransactionAggregationKey(event model.APMEvent, interva
 
 // makeMetricset makes a metricset event from key, counts, and values, with timestamp ts.
 func makeMetricset(key transactionAggregationKey, totalCount int64, counts []int64, values []float64) model.APMEvent {
+	eventOutcomeNumeric := &model.SummaryMetric{
+		Count: 0,
+		Sum:   0,
+	}
+	switch key.eventOutcome {
+	case "success":
+		eventOutcomeNumeric.Count++ // TODO(carsonip): use totalCount?
+		eventOutcomeNumeric.Sum++
+	case "failure":
+		eventOutcomeNumeric.Count++ // TODO(carsonip): use totalCount?
+	case "unknown":
+	}
 	return model.APMEvent{
 		Timestamp:  key.timestamp,
 		Agent:      model.Agent{Name: key.agentName},
@@ -428,7 +440,8 @@ func makeMetricset(key transactionAggregationKey, totalCount int64, counts []int
 			},
 		},
 		Event: model.Event{
-			Outcome: key.eventOutcome,
+			Outcome:        key.eventOutcome,
+			OutcomeNumeric: eventOutcomeNumeric,
 		},
 		FAAS: model.FAAS{
 			Coldstart:   key.faasColdstart,