model: enhancements to support transaction aggregations

beater/config/api_key.go

@@ -62,7 +62,7 @@ func defaultAPIKeyConfig() *APIKeyConfig {
 func (c *APIKeyConfig) Unpack(inp *common.Config) error {
 	cfg := tmpAPIKeyConfig(*defaultAPIKeyConfig())
 	if err := inp.Unpack(&cfg); err != nil {
-		return errors.Errorf("error unpacking api_key config: %w", err)
+		return errors.Wrap(err, "error unpacking api_key config")
 	}
 	*c = APIKeyConfig(cfg)
 	if inp.HasField("elasticsearch") {

beater/config/rum.go

@@ -146,7 +146,7 @@ func (s *SourceMapping) Unpack(inp *common.Config) error {
 
 	cfg := tmpSourceMapping(*defaultSourcemapping())
 	if err := inp.Unpack(&cfg); err != nil {
-		return errors.Errorf("error unpacking sourcemapping config: %w", err)
+		return errors.Wrap(err, "error unpacking sourcemapping config")
 	}
 	*s = SourceMapping(cfg)
 	if inp.HasField("elasticsearch") {

model/metadata/system.go

@@ -38,10 +38,11 @@ func (s *System) name() string {
 	if s.ConfiguredHostname != "" {
 		return s.ConfiguredHostname
 	}
-	return s.hostname()
+	return s.Hostname()
 }
 
-func (s *System) hostname() string {
+// Hostname returns the value to store in `host.hostname`.
+func (s *System) Hostname() string {
 	if s == nil {
 		return ""
 	}
@@ -66,7 +67,7 @@ func (s *System) fields() common.MapStr {
 		return nil
 	}
 	var system mapStr
-	system.maybeSetString("hostname", s.hostname())
+	system.maybeSetString("hostname", s.Hostname())
 	system.maybeSetString("name", s.name())
 	system.maybeSetString("architecture", s.Architecture)
 	if s.Platform != "" {

model/metricset/event.go

@@ -45,50 +45,81 @@ var (
 	processorEntry  = common.MapStr{"name": processorName, "event": docType}
 )
 
+// Metricset describes a set of metrics and associated metadata.
+type Metricset struct {
+	// Timestamp holds the time at which the metrics were published.
+	Timestamp time.Time
+
+	// Metadata holds common metadata describing the entities with which
+	// the metrics are associated: service, system, etc.
+	Metadata metadata.Metadata
+
+	// Transaction holds information about the transaction group with
+	// which the metrics are associated.
+	Transaction Transaction
+
+	// Span holds information about the span types with which the
+	// metrics are associated.
+	Span Span
+
+	// Labels holds arbitrary labels to apply to the metrics.
+	//
+	// These labels override any with the same names in Metadata.Labels.
+	Labels common.MapStr
+
+	// Samples holds the metrics in the set.
+	Samples []Sample
+}
+
+// Sample represents a single named metric.
 type Sample struct {
-	Name  string
+	// Name holds the metric name.
+	Name string
+
+	// Value holds the metric value for single-value metrics.
+	//
+	// If Counts and Values are specified, then Value will be ignored.
 	Value float64
+
+	// Values holds the bucket values for histogram metrics.
+	//
+	// These values must be provided in ascending order.
+	Values []float64
+
+	// Counts holds the bucket counts for histogram metrics.
+	//
+	// These numbers must be positive or zero.
+	//
+	// If Counts is specified, then Values is expected to be
+	// specified with the same number of elements, and with the
+	// same order.
+	Counts []int64
 }
 
-// Transaction provides enough information to connect a metricset to the related kind of transactions
+// Transaction provides enough information to connect a metricset to the related kind of transactions.
 type Transaction struct {
-	Name *string
-	Type *string
-}
+	// Name holds the transaction name: "GET /foo", etc.
+	Name string
 
-// Span provides enough information to connect a metricset to the related kind of spans
-type Span struct {
-	Type    *string
-	Subtype *string
-}
+	// Type holds the transaction type: "request", "message", etc.
+	Type string
 
-type Metricset struct {
-	Metadata    metadata.Metadata
-	Samples     []*Sample
-	Labels      common.MapStr
-	Transaction *Transaction
-	Span        *Span
-	Timestamp   time.Time
-}
+	// Result holds the transaction result: "HTTP 2xx", "OK", "Error", etc.
+	Result string
 
-func (s *Span) fields() common.MapStr {
-	if s == nil {
-		return nil
-	}
-	fields := common.MapStr{}
-	utility.Set(fields, "type", s.Type)
-	utility.Set(fields, "subtype", s.Subtype)
-	return fields
+	// Root indicates whether or not the transaction is the trace root.
+	//
+	// If Root is false, then it will be omitted from the output event.
+	Root bool
 }
 
-func (t *Transaction) fields() common.MapStr {
-	if t == nil {
-		return nil
-	}
-	fields := common.MapStr{}
-	utility.Set(fields, "type", t.Type)
-	utility.Set(fields, "name", t.Name)
-	return fields
+// Span provides enough information to connect a metricset to the related kind of spans.
+type Span struct {
+	// Type holds the span type: "external", "db", etc.
+	Type string
+
+	// Subtype holds the span subtype: "http", "sql", etc.
+	Subtype string
 }
 
 func (me *Metricset) Transform(ctx context.Context, tctx *transform.Context) []beat.Event {
@@ -99,24 +130,58 @@ func (me *Metricset) Transform(ctx context.Context, tctx *transform.Context) []b
 
 	fields := common.MapStr{}
 	for _, sample := range me.Samples {
-		if _, err := fields.Put(sample.Name, sample.Value); err != nil {
+		if err := sample.set(fields); err != nil {
 			logp.NewLogger(logs.Transform).Warnf("failed to transform sample %#v", sample)
 			continue
 		}
 	}
 
 	fields["processor"] = processorEntry
 	me.Metadata.Set(fields)
+	if transactionFields := me.Transaction.fields(); transactionFields != nil {
+		utility.DeepUpdate(fields, transactionKey, transactionFields)
+	}
+	if spanFields := me.Span.fields(); spanFields != nil {
+		utility.DeepUpdate(fields, spanKey, spanFields)
+	}
 
 	// merges with metadata labels, overrides conflicting keys
 	utility.DeepUpdate(fields, "labels", me.Labels)
-	utility.DeepUpdate(fields, transactionKey, me.Transaction.fields())
-	utility.DeepUpdate(fields, spanKey, me.Span.fields())
-
-	return []beat.Event{
-		{
-			Fields:    fields,
-			Timestamp: me.Timestamp,
-		},
+
+	return []beat.Event{{
+		Fields:    fields,
+		Timestamp: me.Timestamp,
+	}}
+}
+
+func (t *Transaction) fields() common.MapStr {
+	var fields mapStr
+	fields.maybeSetString("type", t.Type)
+	fields.maybeSetString("name", t.Name)
+	fields.maybeSetString("result", t.Result)
+	if t.Root {
+		fields.set("root", true)
+	}
+	return common.MapStr(fields)
+}
+
+func (s *Span) fields() common.MapStr {
+	var fields mapStr
+	fields.maybeSetString("type", s.Type)
+	fields.maybeSetString("subtype", s.Subtype)
+	return common.MapStr(fields)
+}
+
+func (s *Sample) set(fields common.MapStr) error {
+	switch {
+	case len(s.Counts) > 0:
+		_, err := fields.Put(s.Name, common.MapStr{
+			"counts": s.Counts,
+			"values": s.Values,
+		})
+		return err
+	default:
+		_, err := fields.Put(s.Name, s.Value)
+		return err
 	}
 }

model/metricset/event_test.go

@@ -36,7 +36,15 @@ func TestTransform(t *testing.T) {
 	metadata := metadata.Metadata{
 		Service: metadata.Service{Name: "myservice"},
 	}
-	spType, spSubtype, trType, trName := "db", "sql", "request", "GET /"
+
+	const (
+		trType   = "request"
+		trName   = "GET /"
+		trResult = "HTTP 2xx"
+
+		spType    = "db"
+		spSubtype = "sql"
+	)
 
 	tests := []struct {
 		Metricset *Metricset
@@ -60,12 +68,48 @@ func TestTransform(t *testing.T) {
 			},
 			Msg: "Payload with empty metric.",
 		},
+		{
+			Metricset: &Metricset{
+				Metadata:  metadata,
+				Timestamp: timestamp,
+				Samples: []Sample{{
+					Name:   "transaction.duration.histogram",
+					Counts: []int64{1},
+					Values: []float64{42},
+				}},
+				Transaction: Transaction{
+					Type:   trType,
+					Name:   trName,
+					Result: trResult,
+					Root:   true,
+				},
+			},
+			Output: []common.MapStr{
+				{
+					"processor": common.MapStr{"event": "metric", "name": "metric"},
+					"service":   common.MapStr{"name": "myservice"},
+					"transaction": common.MapStr{
+						"name":   trName,
+						"type":   trType,
+						"result": trResult,
+						"root":   true,
+						"duration": common.MapStr{
+							"histogram": common.MapStr{
+								"counts": []int64{1},
+								"values": []float64{42},
+							},
+						},
+					},
+				},
+			},
+			Msg: "Payload with extended transaction metadata.",
+		},
 		{
 			Metricset: &Metricset{
 				Metadata:  metadata,
 				Labels:    common.MapStr{"a.b": "a.b.value"},
 				Timestamp: timestamp,
-				Samples: []*Sample{
+				Samples: []Sample{
 					{
 						Name:  "a.counter",
 						Value: 612,
@@ -74,24 +118,32 @@ func TestTransform(t *testing.T) {
 						Name:  "some.gauge",
 						Value: 9.16,
 					},
+					{
+						Name:   "histo.gram",
+						Value:  666, // Value is ignored when Counts/Values are specified
+						Counts: []int64{1, 2, 3},
+						Values: []float64{4.5, 6.0, 9.0},
+					},
 				},
-				Span:        &Span{Type: &spType, Subtype: &spSubtype},
-				Transaction: &Transaction{Type: &trType, Name: &trName},
+				Span:        Span{Type: spType, Subtype: spSubtype},
+				Transaction: Transaction{Type: trType, Name: trName},
 			},
 			Output: []common.MapStr{
 				{
-					"labels": common.MapStr{
-						"a.b": "a.b.value",
-					},
-					"service": common.MapStr{
-						"name": "myservice",
-					},
-
-					"a":           common.MapStr{"counter": float64(612)},
-					"some":        common.MapStr{"gauge": float64(9.16)},
 					"processor":   common.MapStr{"event": "metric", "name": "metric"},
+					"service":     common.MapStr{"name": "myservice"},
 					"transaction": common.MapStr{"name": trName, "type": trType},
 					"span":        common.MapStr{"type": spType, "subtype": spSubtype},
+					"labels":      common.MapStr{"a.b": "a.b.value"},
+
+					"a":    common.MapStr{"counter": float64(612)},
+					"some": common.MapStr{"gauge": float64(9.16)},
+					"histo": common.MapStr{
+						"gram": common.MapStr{
+							"counts": []int64{1, 2, 3},
+							"values": []float64{4.5, 6.0, 9.0},
+						},
+					},
 				},
 			},
 			Msg: "Payload with valid metric.",

model/metricset/mapstr.go

@@ -0,0 +1,37 @@
+// Licensed to Elasticsearch B.V. under one or more contributor
+// license agreements. See the NOTICE file distributed with
+// this work for additional information regarding copyright
+// ownership. Elasticsearch B.V. licenses this file to you under
+// the Apache License, Version 2.0 (the "License"); you may
+// not use this file except in compliance with the License.
+// You may obtain a copy of the License at
+//
+//     http://www.apache.org/licenses/LICENSE-2.0
+//
+// Unless required by applicable law or agreed to in writing,
+// software distributed under the License is distributed on an
+// "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
+// KIND, either express or implied.  See the License for the
+// specific language governing permissions and limitations
+// under the License.
+
+package metricset
+
+import "github.com/elastic/beats/v7/libbeat/common"
+
+type mapStr common.MapStr
+
+func (m *mapStr) set(k string, v interface{}) {
+	if *m == nil {
+		*m = make(mapStr)
+	}
+	(*m)[k] = v
+}
+
+func (m *mapStr) maybeSetString(k, v string) bool {
+	if v != "" {
+		m.set(k, v)
+		return true
+	}
+	return false
+}