Support composite spans in the intake API

beater/test_approved_es_documents/TestPublishIntegrationSpans.approved.json

@@ -893,6 +893,146 @@
                 "id": "123",
                 "name": "john"
             }
+        },
+        {
+            "@timestamp": "2021-07-06T11:58:05.682Z",
+            "agent": {
+                "name": "elastic-node",
+                "version": "3.14.0"
+            },
+            "cloud": {
+                "account": {
+                    "id": "account_id",
+                    "name": "account_name"
+                },
+                "availability_zone": "cloud_availability_zone",
+                "instance": {
+                    "id": "instance_id",
+                    "name": "instance_name"
+                },
+                "machine": {
+                    "type": "machine_type"
+                },
+                "project": {
+                    "id": "project_id",
+                    "name": "project_name"
+                },
+                "provider": "cloud_provider",
+                "region": "cloud_region",
+                "service": {
+                    "name": "lambda"
+                }
+            },
+            "container": {
+                "id": "container-id"
+            },
+            "ecs": {
+                "version": "1.10.0"
+            },
+            "event": {
+                "outcome": "success"
+            },
+            "host": {
+                "architecture": "x64",
+                "hostname": "node-name",
+                "ip": "127.0.0.1",
+                "name": "node-name",
+                "os": {
+                    "platform": "darwin"
+                }
+            },
+            "kubernetes": {
+                "namespace": "namespace1",
+                "node": {
+                    "name": "node-name"
+                },
+                "pod": {
+                    "name": "pod-name",
+                    "uid": "pod-uid"
+                }
+            },
+            "labels": {
+                "tag1": "label1"
+            },
+            "observer": {
+                "ephemeral_id": "00000000-0000-0000-0000-000000000000",
+                "hostname": "",
+                "id": "fbba762a-14dd-412c-b7e9-b79f903eb492",
+                "type": "test-apm-server",
+                "version": "1.2.3",
+                "version_major": 1
+            },
+            "parent": {
+                "id": "abcdef0123456789"
+            },
+            "process": {
+                "args": [
+                    "node",
+                    "server.js"
+                ],
+                "pid": 1234,
+                "ppid": 6789,
+                "title": "node"
+            },
+            "processor": {
+                "event": "span",
+                "name": "transaction"
+            },
+            "service": {
+                "environment": "staging",
+                "framework": {
+                    "name": "Express",
+                    "version": "1.2.3"
+                },
+                "language": {
+                    "name": "ecmascript",
+                    "version": "8"
+                },
+                "name": "backendspans",
+                "node": {
+                    "name": "container-id"
+                },
+                "runtime": {
+                    "name": "node",
+                    "version": "8.0.0"
+                },
+                "version": "5.1.3"
+            },
+            "span": {
+                "action": "query",
+                "composite": {
+                    "compression_strategy": "exact_match",
+                    "count": 10,
+                    "sum": {
+                        "us": 32592
+                    }
+                },
+                "duration": {
+                    "us": 3781
+                },
+                "id": "abcdef01234567",
+                "name": "SELECT FROM p_details",
+                "start": {
+                    "us": 2830
+                },
+                "subtype": "postgresql",
+                "type": "db"
+            },
+            "timestamp": {
+                "us": 1625572685682272
+            },
+            "trace": {
+                "id": "edcbaf0123456789abcdef9876543210"
+            },
+            "transaction": {
+                "id": "01af25874dec69dd"
+            },
+            "user": {
+                "domain": "ldap://abc",
+                "email": "s@test.com",
+                "id": "123",
+                "name": "john"
+            }
         }
     ]
 }

docs/spec/v2/span.json

@@ -22,6 +22,34 @@
       },
       "minItems": 0
     },
+    "composite": {
+      "description": "Composite holds details on a group of spans represented by a single one.",
+      "type": [
+        "null",
+        "object"
+      ],
+      "properties": {
+        "compression_strategy": {
+          "description": "A string value indicating which compression strategy was used. The valid values are `exact_match` and `same_kind`.",
+          "type": "string"
+        },
+        "count": {
+          "description": "Count is the number of compressed spans the composite span represents. The minimum count is 2, as a composite span represents at least two spans.",
+          "type": "integer",
+          "minimum": 2
+        },
+        "sum": {
+          "description": "Sum is the durations of all compressed spans this composite span represents in milliseconds. Thus sum is the net duration of all the compressed spans while duration is the gross duration, including \"whitespace\" between the spans.",
+          "type": "number",
+          "minimum": 0
+        }
+      },
+      "required": [
+        "count",
+        "sum",
+        "compression_strategy"
+      ]
+    },
     "context": {
       "description": "Context holds arbitrary contextual information for the event.",
       "type": [

model/modeldecoder/rumv3/transaction_test.go

@@ -223,6 +223,7 @@ func TestDecodeMapToTransactionModel(t *testing.T) {
 				"Metadata",
 				// values not set for RUM v3
 				"ChildIDs",
+				"Composite",
 				"DB",
 				"Experimental",
 				"HTTP.Response.Headers",

model/modeldecoder/v2/decoder.go

@@ -770,6 +770,19 @@ func mapToSpanModel(from *span, metadata *model.Metadata, reqTime time.Time, con
 			out.Type = from.Type.Val
 		}
 	}
+	if from.Composite.IsSet() {
+		composite := model.Composite{}
+		if from.Composite.Count.IsSet() {
+			composite.Count = from.Composite.Count.Val
+		}
+		if from.Composite.Sum.IsSet() {
+			composite.Sum = from.Composite.Sum.Val
+		}
+		if from.Composite.CompressionStrategy.IsSet() {
+			composite.CompressionStrategy = from.Composite.CompressionStrategy.Val
+		}
+		out.Composite = &composite
+	}
 	if len(from.ChildIDs) > 0 {
 		out.ChildIDs = make([]string, len(from.ChildIDs))
 		copy(out.ChildIDs, from.ChildIDs)

model/modeldecoder/v2/model.go

@@ -605,6 +605,8 @@ type span struct {
 	Action nullable.String `json:"action" validate:"maxLength=1024"`
 	// ChildIDs holds a list of successor transactions and/or spans.
 	ChildIDs []string `json:"child_ids" validate:"maxLength=1024"`
+	// Composite holds details on a group of spans represented by a single one.
+	Composite spanComposite `json:"composite"`
 	// Context holds arbitrary contextual information for the event.
 	Context spanContext `json:"context"`
 	// Duration of the span in milliseconds
@@ -768,6 +770,20 @@ type stacktraceFrame struct {
 	_    struct{}      `validate:"requiredAnyOf=classname;filename"`
 }
 
+type spanComposite struct {
+	// Count is the number of compressed spans the composite span represents.
+	// The minimum count is 2, as a composite span represents at least two spans.
+	Count nullable.Int `json:"count" validate:"required,min=2"`
+	// Sum is the durations of all compressed spans this composite span
+	// represents in milliseconds. Thus sum is the net duration of all the
+	// compressed spans while duration is the gross duration, including
+	// "whitespace" between the spans.
+	Sum nullable.Float64 `json:"sum" validate:"required,min=0"`
+	// A string value indicating which compression strategy was used. The valid
+	// values are `exact_match` and `same_kind`.
+	CompressionStrategy nullable.String `json:"compression_strategy" validate:"required"`
+}
+
 type transaction struct {
 	// Context holds arbitrary contextual information for the event.
 	Context context `json:"context"`

model/modeldecoder/v2/model_test.go

@@ -508,6 +508,8 @@ func TestSpanRequiredValidationRules(t *testing.T) {
 	var event span
 	modeldecodertest.InitStructValues(&event)
 	event.Outcome.Set("failure")
+	// Composite.Count must be > 1
+	event.Composite.Count.Set(2)
 	// test vanilla struct is valid
 	require.NoError(t, event.validate())
 
@@ -521,6 +523,9 @@ func TestSpanRequiredValidationRules(t *testing.T) {
 		"parent_id":                            nil,
 		"trace_id":                             nil,
 		"type":                                 nil,
+		"composite.count":                      nil,
+		"composite.sum":                        nil,
+		"composite.compression_strategy":       nil,
 	}
 	cb := assertRequiredFn(t, requiredKeys, event.validate)
 	modeldecodertest.SetZeroStructValue(&event, cb)

model/span.go

@@ -68,6 +68,7 @@ type Span struct {
 	URL                string
 	Destination        *Destination
 	DestinationService *DestinationService
+	Composite          *Composite
 
 	Experimental interface{}
 
@@ -102,6 +103,13 @@ type DestinationService struct {
 	Resource string
 }
 
+// Composite holds details on a group of spans compressed into one.
+type Composite struct {
+	Count               int
+	Sum                 float64
+	CompressionStrategy string
+}
+
 func (db *DB) fields() common.MapStr {
 	if db == nil {
 		return nil
@@ -143,6 +151,18 @@ func (d *DestinationService) fields() common.MapStr {
 	return common.MapStr(fields)
 }
 
+func (c *Composite) fields() common.MapStr {
+	if c == nil {
+		return nil
+	}
+	var fields mapStr
+	fields.set("count", c.Count)
+	fields.set("sum", utility.MillisAsMicros(c.Sum))
+	fields.set("compression_strategy", c.CompressionStrategy)
+
+	return common.MapStr(fields)
+}
+
 func (e *Span) toBeatEvent(ctx context.Context) beat.Event {
 	spanTransformations.Inc()
 	if frames := len(e.Stacktrace); frames > 0 {
@@ -214,6 +234,7 @@ func (e *Span) fields(ctx context.Context) common.MapStr {
 	}
 	fields.maybeSetMapStr("db", e.DB.fields())
 	fields.maybeSetMapStr("message", e.Message.Fields())
+	fields.maybeSetMapStr("composite", e.Composite.fields())
 	if destinationServiceFields := e.DestinationService.fields(); len(destinationServiceFields) > 0 {
 		common.MapStr(fields).Put("destination.service", destinationServiceFields)
 	}

model/span_test.go

@@ -118,7 +118,8 @@ func TestSpanTransform(t *testing.T) {
 					Name:     destServiceName,
 					Resource: destServiceResource,
 				},
-				Message: &Message{QueueName: "users"},
+				Message:   &Message{QueueName: "users"},
+				Composite: &Composite{Count: 10, Sum: 1.1, CompressionStrategy: "exact_match"},
 			},
 			Output: common.MapStr{
 				"span": common.MapStr{
@@ -153,6 +154,11 @@ func TestSpanTransform(t *testing.T) {
 						},
 					},
 					"message": common.MapStr{"queue": common.MapStr{"name": "users"}},
+					"composite": common.MapStr{
+						"count":                10,
+						"sum":                  common.MapStr{"us": 1100},
+						"compression_strategy": "exact_match",
+					},
 				},
 				"labels":      common.MapStr{"label_a": 12, "label_b": "b", "c": 1},
 				"processor":   common.MapStr{"event": "span", "name": "transaction"},