Introduce model.Processor (#5984) (#6033)

* Introduce model.Processor Processor identifies the event kind: transaction, span, metricset, error, or profile. This is used for routing events to the appropriate index or data stream, and in queries for fetching events of a given kind. * Update SetDataStream to use APMEvent.Processor * Move monitoring event counters to model processor Also, remove unused metrics. * Update aggregators to use Processor * sampling: use Processor * Update changelog * Remove redundant setting of event.Processor Co-authored-by: stuart nelson <[email protected]> # Conflicts: # changelogs/head.asciidoc
elastic · Aug 24, 2021 · 11ea652 · 11ea652
1 parent 664bbf3
commit 11ea652
Show file tree

Hide file tree

Showing 34 changed files with 408 additions and 238 deletions.
diff --git a/beater/api/profile/convert.go b/beater/api/profile/convert.go
@@ -87,6 +87,7 @@ func appendProfileSampleBatch(pp *profile.Profile, baseEvent model.APMEvent, out
 		}
 
 		event := baseEvent
+		event.Processor = model.ProfileProcessor
 		event.Labels = event.Labels.Clone()
 		if n := len(sample.Label); n > 0 {
 			for k, v := range sample.Label {

diff --git a/beater/beater.go b/beater/beater.go
@@ -30,6 +30,7 @@ import (
 
 	"github.com/elastic/beats/v7/libbeat/common/transport"
 	"github.com/elastic/beats/v7/libbeat/common/transport/tlscommon"
+	"github.com/elastic/beats/v7/libbeat/monitoring"
 	"github.com/elastic/go-ucfg"
 
 	"github.com/pkg/errors"
@@ -479,6 +480,7 @@ func (s *serverRunner) wrapRunServerWithPreprocessors(runServer RunServerFunc) R
 		modelprocessor.SetErrorMessage{},
 		newObserverBatchProcessor(s.beat.Info),
 		model.ProcessBatchFunc(ecsVersionBatchProcessor),
+		modelprocessor.NewEventCounter(monitoring.Default.GetRegistry("apm-server")),
 	}
 	if s.config.DefaultServiceEnvironment != "" {
 		processors = append(processors, &modelprocessor.SetDefaultServiceEnvironment{

diff --git a/model/apmevent.go b/model/apmevent.go
@@ -53,6 +53,7 @@ type APMEvent struct {
 	Network     Network
 	Session     Session
 	URL         URL
+	Processor   Processor
 	Trace       Trace
 
 	// Timestamp holds the event timestamp.
@@ -91,7 +92,8 @@ func (e *APMEvent) BeatEvent(ctx context.Context) beat.Event {
 		event.Fields = e.Error.fields()
 	case e.ProfileSample != nil:
 		event.Fields = e.ProfileSample.fields()
-	default:
+	}
+	if event.Fields == nil {
 		event.Fields = make(common.MapStr)
 	}
 
@@ -133,6 +135,7 @@ func (e *APMEvent) BeatEvent(ctx context.Context) beat.Event {
 	fields.maybeSetMapStr("event", e.Event.fields())
 	fields.maybeSetMapStr("url", e.URL.fields())
 	fields.maybeSetMapStr("session", e.Session.fields())
+	fields.maybeSetMapStr("processor", e.Processor.fields())
 	fields.maybeSetMapStr("trace", e.Trace.fields())
 	fields.maybeSetString("message", e.Message)
 	return event

diff --git a/model/apmevent_test.go b/model/apmevent_test.go
@@ -74,6 +74,7 @@ func TestAPMEventFields(t *testing.T) {
 				Message:     "bottle",
 				Transaction: &Transaction{},
 				Timestamp:   time.Date(2019, 1, 3, 15, 17, 4, 908.596*1e6, time.FixedZone("+0100", 3600)),
+				Processor:   Processor{Name: "processor_name", Event: "processor_event"},
 				Trace:       Trace{ID: traceID},
 			},
 			output: common.MapStr{
@@ -107,12 +108,12 @@ func TestAPMEventFields(t *testing.T) {
 				"trace": common.MapStr{
 					"id": traceID,
 				},
-
-				// fields related to APMEvent.Transaction
 				"processor": common.MapStr{
-					"name":  "transaction",
-					"event": "transaction",
+					"name":  "processor_name",
+					"event": "processor_event",
 				},
+
+				// fields related to APMEvent.Transaction
 				"timestamp": common.MapStr{"us": int64(1546525024908596)},
 				"transaction": common.MapStr{
 					"duration": common.MapStr{"us": 0},

diff --git a/model/error.go b/model/error.go
@@ -19,21 +19,15 @@ package model
 
 import (
 	"github.com/elastic/beats/v7/libbeat/common"
-	"github.com/elastic/beats/v7/libbeat/monitoring"
 )
 
 var (
-	errorMetrics           = monitoring.Default.NewRegistry("apm-server.processor.error")
-	errorTransformations   = monitoring.NewInt(errorMetrics, "transformations")
-	errorStacktraceCounter = monitoring.NewInt(errorMetrics, "stacktraces")
-	errorFrameCounter      = monitoring.NewInt(errorMetrics, "frames")
-	errorProcessorEntry    = common.MapStr{"name": errorProcessorName, "event": errorDocType}
+	// ErrorProcessor is the Processor value that should be assigned to error events.
+	ErrorProcessor = Processor{Name: "error", Event: "error"}
 )
 
 const (
-	errorProcessorName = "error"
-	errorDocType       = "error"
-	ErrorsDataset      = "apm.error"
+	ErrorsDataset = "apm.error"
 )
 
 type Error struct {
@@ -76,16 +70,7 @@ type Log struct {
 }
 
 func (e *Error) fields() common.MapStr {
-	errorTransformations.Inc()
-
-	if e.Exception != nil {
-		addStacktraceCounter(e.Exception.Stacktrace)
-	}
-	if e.Log != nil {
-		addStacktraceCounter(e.Log.Stacktrace)
-	}
-
-	fields := mapStr{"processor": errorProcessorEntry}
+	var fields mapStr
 	if e.HTTP != nil {
 		fields.maybeSetMapStr("http", e.HTTP.transactionTopLevelFields())
 	}
@@ -161,13 +146,6 @@ func (e *Error) logFields() common.MapStr {
 	return common.MapStr(log)
 }
 
-func addStacktraceCounter(st Stacktrace) {
-	if frames := len(st); frames > 0 {
-		errorStacktraceCounter.Inc()
-		errorFrameCounter.Add(int64(frames))
-	}
-}
-
 // flattenExceptionTree recursively traverses the causes of an exception to return a slice of exceptions.
 // Tree traversal is Depth First.
 // The parent of a exception in the resulting slice is at the position indicated by the `parent` property

diff --git a/model/metricset.go b/model/metricset.go
@@ -19,23 +19,16 @@ package model
 
 import (
 	"github.com/elastic/beats/v7/libbeat/common"
-	"github.com/elastic/beats/v7/libbeat/monitoring"
 )
 
 const (
-	metricsetProcessorName  = "metric"
-	metricsetDocType        = "metric"
-	metricsetEventKey       = "event"
-	metricsetTransactionKey = "transaction"
-	metricsetSpanKey        = "span"
-	AppMetricsDataset       = "apm.app"
-	InternalMetricsDataset  = "apm.internal"
+	AppMetricsDataset      = "apm.app"
+	InternalMetricsDataset = "apm.internal"
 )
 
 var (
-	metricsetMetrics         = monitoring.Default.NewRegistry("apm-server.processor.metric")
-	metricsetTransformations = monitoring.NewInt(metricsetMetrics, "transformations")
-	metricsetProcessorEntry  = common.MapStr{"name": metricsetProcessorName, "event": metricsetDocType}
+	// MetricsetProcessor is the Processor value that should be assigned to metricset events.
+	MetricsetProcessor = Processor{Name: "metric", Event: "metric"}
 )
 
 // MetricType describes the type of a metric: gauge, counter, or histogram.
@@ -141,13 +134,9 @@ type MetricsetSpan struct {
 }
 
 func (me *Metricset) fields() common.MapStr {
-	metricsetTransformations.Inc()
-
 	var fields mapStr
-	fields.set("processor", metricsetProcessorEntry)
-
-	fields.maybeSetMapStr(metricsetTransactionKey, me.Transaction.fields())
-	fields.maybeSetMapStr(metricsetSpanKey, me.Span.fields())
+	fields.maybeSetMapStr("transaction", me.Transaction.fields())
+	fields.maybeSetMapStr("span", me.Span.fields())
 	if me.TimeseriesInstanceID != "" {
 		fields.set("timeseries", common.MapStr{"instance": me.TimeseriesInstanceID})
 	}
@@ -158,7 +147,7 @@ func (me *Metricset) fields() common.MapStr {
 
 	var metricDescriptions mapStr
 	for name, sample := range me.Samples {
-		sample.set(name, fields)
+		sample.set(name, &fields)
 
 		var md mapStr
 		md.maybeSetString("type", string(sample.Type))
@@ -190,7 +179,7 @@ func (s *MetricsetSpan) fields() common.MapStr {
 	return common.MapStr(fields)
 }
 
-func (s *MetricsetSample) set(name string, fields mapStr) {
+func (s *MetricsetSample) set(name string, fields *mapStr) {
 	if s.Type == MetricTypeHistogram {
 		fields.set(name, common.MapStr{
 			"counts": s.Counts,

diff --git a/model/metricset_test.go b/model/metricset_test.go
@@ -46,15 +46,12 @@ func TestMetricset(t *testing.T) {
 	}{
 		{
 			Metricset: &Metricset{},
-			Output: common.MapStr{
-				"processor": common.MapStr{"event": "metric", "name": "metric"},
-			},
-			Msg: "Payload with empty metric.",
+			Output:    common.MapStr{},
+			Msg:       "Payload with empty metric.",
 		},
 		{
 			Metricset: &Metricset{Name: "raj"},
 			Output: common.MapStr{
-				"processor":      common.MapStr{"event": "metric", "name": "metric"},
 				"metricset.name": "raj",
 			},
 			Msg: "Payload with metricset name.",
@@ -67,7 +64,6 @@ func TestMetricset(t *testing.T) {
 				},
 			},
 			Output: common.MapStr{
-				"processor":  common.MapStr{"event": "metric", "name": "metric"},
 				"a.counter":  612.0,
 				"some.gauge": 9.16,
 			},
@@ -82,7 +78,6 @@ func TestMetricset(t *testing.T) {
 				},
 			},
 			Output: common.MapStr{
-				"processor":   common.MapStr{"event": "metric", "name": "metric"},
 				"transaction": common.MapStr{"type": trType, "name": trName},
 				"span": common.MapStr{
 					"type": spType, "subtype": spSubtype,
@@ -111,7 +106,6 @@ func TestMetricset(t *testing.T) {
 				DocCount: 6,
 			},
 			Output: common.MapStr{
-				"processor":  common.MapStr{"event": "metric", "name": "metric"},
 				"timeseries": common.MapStr{"instance": "foo"},
 				"transaction": common.MapStr{
 					"type":   trType,
@@ -143,7 +137,6 @@ func TestMetricset(t *testing.T) {
 				},
 			},
 			Output: common.MapStr{
-				"processor": common.MapStr{"event": "metric", "name": "metric"},
 				"span": common.MapStr{
 					"type": spType, "subtype": spSubtype,
 					"destination": common.MapStr{"service": common.MapStr{"resource": resource}},
@@ -173,7 +166,6 @@ func TestMetricset(t *testing.T) {
 				},
 			},
 			Output: common.MapStr{
-				"processor": common.MapStr{"event": "metric", "name": "metric"},
 				"latency_histogram": common.MapStr{
 					"counts": []int64{1, 2, 3},
 					"values": []float64{1.1, 2.2, 3.3},

diff --git a/model/modeldecoder/rumv3/decoder.go b/model/modeldecoder/rumv3/decoder.go
@@ -161,20 +161,20 @@ func DecodeNestedTransaction(d decoder.Decoder, input *modeldecoder.Input, batch
 	*batch = append(*batch, transaction)
 
 	for _, m := range root.Transaction.Metricsets {
-		metricset := input.Base
-		mapToMetricsetModel(&m, &metricset)
-		metricset.Metricset.Transaction.Name = transaction.Transaction.Name
-		metricset.Metricset.Transaction.Type = transaction.Transaction.Type
-		*batch = append(*batch, metricset)
+		event := input.Base
+		mapToMetricsetModel(&m, &event)
+		event.Metricset.Transaction.Name = transaction.Transaction.Name
+		event.Metricset.Transaction.Type = transaction.Transaction.Type
+		*batch = append(*batch, event)
 	}
 
 	offset := len(*batch)
 	for _, s := range root.Transaction.Spans {
-		span := input.Base
-		mapToSpanModel(&s, &span)
-		span.Span.TransactionID = transaction.Transaction.ID
-		span.Trace = transaction.Trace
-		*batch = append(*batch, span)
+		event := input.Base
+		mapToSpanModel(&s, &event)
+		event.Span.TransactionID = transaction.Transaction.ID
+		event.Trace = transaction.Trace
+		*batch = append(*batch, event)
 	}
 	spans := (*batch)[offset:]
 	for i, s := range root.Transaction.Spans {
@@ -190,6 +190,7 @@ func DecodeNestedTransaction(d decoder.Decoder, input *modeldecoder.Input, batch
 func mapToErrorModel(from *errorEvent, event *model.APMEvent) {
 	out := &model.Error{}
 	event.Error = out
+	event.Processor = model.ErrorProcessor
 
 	// overwrite metadata with event specific information
 	mapToServiceModel(from.Context.Service, &event.Service)
@@ -391,6 +392,7 @@ func mapToMetadataModel(m *metadata, out *model.APMEvent) {
 func mapToMetricsetModel(from *metricset, event *model.APMEvent) {
 	out := &model.Metricset{}
 	event.Metricset = out
+	event.Processor = model.MetricsetProcessor
 
 	// map samples information
 	if from.Samples.IsSet() {
@@ -512,6 +514,7 @@ func mapToAgentModel(from contextServiceAgent, out *model.Agent) {
 func mapToSpanModel(from *span, event *model.APMEvent) {
 	out := &model.Span{}
 	event.Span = out
+	event.Processor = model.SpanProcessor
 
 	// map span specific data
 	if !from.Action.IsSet() && !from.Subtype.IsSet() {
@@ -690,6 +693,7 @@ func mapToStracktraceModel(from []stacktraceFrame, out model.Stacktrace) {
 func mapToTransactionModel(from *transaction, event *model.APMEvent) {
 	out := &model.Transaction{}
 	event.Transaction = out
+	event.Processor = model.TransactionProcessor
 
 	// overwrite metadata with event specific information
 	mapToServiceModel(from.Context.Service, &event.Service)

diff --git a/model/modeldecoder/rumv3/metadata_test.go b/model/modeldecoder/rumv3/metadata_test.go
@@ -60,6 +60,7 @@ func metadataExceptions(keys ...string) func(key string) bool {
 		"Network",
 		"Observer",
 		"Process",
+		"Processor",
 		"Service.Node",
 		"Service.Agent.EphemeralID",
 		"Host",

diff --git a/model/modeldecoder/v2/decoder.go b/model/modeldecoder/v2/decoder.go
@@ -240,6 +240,7 @@ func mapToClientModel(from contextRequest, out *model.Client) {
 func mapToErrorModel(from *errorEvent, config modeldecoder.Config, event *model.APMEvent) {
 	out := &model.Error{}
 	event.Error = out
+	event.Processor = model.ErrorProcessor
 
 	// overwrite metadata with event specific information
 	mapToServiceModel(from.Context.Service, &event.Service)
@@ -538,6 +539,7 @@ func mapToMetadataModel(from *metadata, out *model.APMEvent) {
 func mapToMetricsetModel(from *metricset, config modeldecoder.Config, event *model.APMEvent) {
 	out := &model.Metricset{}
 	event.Metricset = out
+	event.Processor = model.MetricsetProcessor
 
 	if !from.Timestamp.Val.IsZero() {
 		event.Timestamp = from.Timestamp.Val
@@ -725,6 +727,7 @@ func mapToAgentModel(from contextServiceAgent, out *model.Agent) {
 func mapToSpanModel(from *span, config modeldecoder.Config, event *model.APMEvent) {
 	out := &model.Span{}
 	event.Span = out
+	event.Processor = model.SpanProcessor
 
 	// map span specific data
 	if !from.Action.IsSet() && !from.Subtype.IsSet() {
@@ -987,6 +990,7 @@ func mapToStracktraceModel(from []stacktraceFrame, out model.Stacktrace) {
 
 func mapToTransactionModel(from *transaction, config modeldecoder.Config, event *model.APMEvent) {
 	out := &model.Transaction{}
+	event.Processor = model.TransactionProcessor
 	event.Transaction = out
 
 	// overwrite metadata with event specific information

diff --git a/model/modeldecoder/v2/metadata_test.go b/model/modeldecoder/v2/metadata_test.go
@@ -74,6 +74,9 @@ func isUnmappedMetadataField(key string) bool {
 		"Observer.VersionMajor",
 		"Process.CommandLine",
 		"Process.Executable",
+		"Processor",
+		"Processor.Event",
+		"Processor.Name",
 		"Host.OS.Full",
 		"Host.OS.Type",
 		"Host.ID",

diff --git a/model/modelprocessor/datastream.go b/model/modelprocessor/datastream.go
@@ -39,14 +39,14 @@ func (s *SetDataStream) ProcessBatch(ctx context.Context, b *model.Batch) error
 }
 
 func (s *SetDataStream) setDataStream(event *model.APMEvent) {
-	switch {
-	case event.Transaction != nil || event.Span != nil:
+	switch event.Processor {
+	case model.SpanProcessor, model.TransactionProcessor:
 		event.DataStream.Type = datastreams.TracesType
 		event.DataStream.Dataset = model.TracesDataset
-	case event.Error != nil:
+	case model.ErrorProcessor:
 		event.DataStream.Type = datastreams.LogsType
 		event.DataStream.Dataset = model.ErrorsDataset
-	case event.Metricset != nil:
+	case model.MetricsetProcessor:
 		event.DataStream.Type = datastreams.MetricsType
 		// Metrics that include well-defined transaction/span fields
 		// (i.e. breakdown metrics, transaction and span metrics) will
@@ -58,7 +58,7 @@ func (s *SetDataStream) setDataStream(event *model.APMEvent) {
 				datastreams.NormalizeServiceName(event.Service.Name),
 			)
 		}
-	case event.ProfileSample != nil:
+	case model.ProfileProcessor:
 		event.DataStream.Type = datastreams.MetricsType
 		event.DataStream.Dataset = model.ProfilesDataset
 	}