add service name to dataset field

elastic · Feb 15, 2021 · 5833656 · 5833656
1 parent 9ad7072
commit 5833656
Show file tree

Hide file tree

Showing 43 changed files with 107 additions and 101 deletions.
diff --git a/datastreams/servicename.go b/datastreams/servicename.go
@@ -25,7 +25,6 @@ import "strings"
 // Concretely, this function will lowercase the string and replace any
 // reserved characters with "_".
 //
-// TODO: use when Fleet supports variables in data streams (see #4492)
 func NormalizeServiceName(s string) string {
 	s = strings.ToLower(s)
 	s = strings.Map(replaceReservedRune, s)

diff --git a/model/error.go b/model/error.go
@@ -121,7 +121,8 @@ func (e *Error) Transform(ctx context.Context, cfg *transform.Config) []beat.Eve
 		// By storing errors in a "logs" data stream, they can be viewed in the Logs app
 		// in Kibana.
 		fields[datastreams.TypeField] = datastreams.LogsType
-		fields[datastreams.DatasetField] = ErrorsDataset
+		dataset := fmt.Sprintf("%s.%s", ErrorsDataset, datastreams.NormalizeServiceName(e.Metadata.Service.Name))
+		fields[datastreams.DatasetField] = dataset
 	}
 
 	// first set the generic metadata (order is relevant)

diff --git a/model/error_test.go b/model/error_test.go
@@ -308,7 +308,7 @@ func TestEvents(t *testing.T) {
 			Transformable: &Error{Timestamp: timestamp, Metadata: md},
 			Output: common.MapStr{
 				"data_stream.type":    "logs",
-				"data_stream.dataset": "apm.error",
+				"data_stream.dataset": "apm.error.myservice",
 				"agent":               common.MapStr{"name": "go", "version": "1.0"},
 				"service":             common.MapStr{"name": "myservice", "version": "1.0"},
 				"error": common.MapStr{
@@ -323,7 +323,7 @@ func TestEvents(t *testing.T) {
 			Transformable: &Error{Timestamp: timestamp, Metadata: md, TransactionSampled: &sampledFalse},
 			Output: common.MapStr{
 				"data_stream.type":    "logs",
-				"data_stream.dataset": "apm.error",
+				"data_stream.dataset": "apm.error.myservice",
 				"transaction":         common.MapStr{"sampled": false},
 				"agent":               common.MapStr{"name": "go", "version": "1.0"},
 				"service":             common.MapStr{"name": "myservice", "version": "1.0"},
@@ -339,7 +339,7 @@ func TestEvents(t *testing.T) {
 			Transformable: &Error{Timestamp: timestamp, Metadata: md, TransactionType: &transactionType},
 			Output: common.MapStr{
 				"data_stream.type":    "logs",
-				"data_stream.dataset": "apm.error",
+				"data_stream.dataset": "apm.error.myservice",
 				"transaction":         common.MapStr{"type": "request"},
 				"error": common.MapStr{
 					"grouping_key": "d41d8cd98f00b204e9800998ecf8427e",
@@ -370,7 +370,7 @@ func TestEvents(t *testing.T) {
 
 			Output: common.MapStr{
 				"data_stream.type":    "logs",
-				"data_stream.dataset": "apm.error",
+				"data_stream.dataset": "apm.error.myservice",
 				"labels":              common.MapStr{"key": true, "label": 101},
 				"service":             common.MapStr{"name": "myservice", "version": "1.0"},
 				"agent":               common.MapStr{"name": "go", "version": "1.0"},

diff --git a/model/metricset.go b/model/metricset.go
@@ -199,15 +199,16 @@ func (me *Metricset) Transform(ctx context.Context, cfg *transform.Config) []bea
 	fields["processor"] = metricsetProcessorEntry
 
 	if cfg.DataStreams {
+		dataset := AppMetricsDataset
 		// Metrics are stored in "metrics" data streams.
 		if isInternal {
 			// Metrics that include well-defined transaction/span fields
 			// (i.e. breakdown metrics, transaction and span metrics) will
 			// be stored separately from application and runtime metrics.
-			fields[datastreams.DatasetField] = InternalMetricsDataset
-		} else {
-			fields[datastreams.DatasetField] = AppMetricsDataset
+			dataset = InternalMetricsDataset
 		}
+		dataset += "." + datastreams.NormalizeServiceName(me.Metadata.Service.Name)
+		fields[datastreams.DatasetField] = dataset
 		fields[datastreams.TypeField] = datastreams.MetricsType
 	}
 

diff --git a/model/metricset_test.go b/model/metricset_test.go
@@ -63,7 +63,7 @@ func TestTransform(t *testing.T) {
 			Output: []common.MapStr{
 				{
 					"data_stream.type":    "metrics",
-					"data_stream.dataset": "apm.app",
+					"data_stream.dataset": "apm.app.myservice",
 					"processor":           common.MapStr{"event": "metric", "name": "metric"},
 					"service": common.MapStr{
 						"name": "myservice",
@@ -91,7 +91,7 @@ func TestTransform(t *testing.T) {
 			Output: []common.MapStr{
 				{
 					"data_stream.type":    "metrics",
-					"data_stream.dataset": "apm.app",
+					"data_stream.dataset": "apm.app.myservice",
 					"processor":           common.MapStr{"event": "metric", "name": "metric"},
 					"service":             common.MapStr{"name": "myservice"},
 					"labels":              common.MapStr{"a_b": "a.b.value"},
@@ -116,7 +116,7 @@ func TestTransform(t *testing.T) {
 			Output: []common.MapStr{
 				{
 					"data_stream.type":    "metrics",
-					"data_stream.dataset": "apm.internal",
+					"data_stream.dataset": "apm.internal.myservice",
 					"processor":           common.MapStr{"event": "metric", "name": "metric"},
 					"service":             common.MapStr{"name": "myservice"},
 					"transaction":         common.MapStr{"type": trType, "name": trName},
@@ -154,7 +154,7 @@ func TestTransform(t *testing.T) {
 			Output: []common.MapStr{
 				{
 					"data_stream.type":    "metrics",
-					"data_stream.dataset": "apm.internal",
+					"data_stream.dataset": "apm.internal.myservice",
 					"processor":           common.MapStr{"event": "metric", "name": "metric"},
 					"service":             common.MapStr{"name": "myservice"},
 					"event":               common.MapStr{"outcome": eventOutcome},
@@ -197,7 +197,7 @@ func TestTransform(t *testing.T) {
 			Output: []common.MapStr{
 				{
 					"data_stream.type":    "metrics",
-					"data_stream.dataset": "apm.internal",
+					"data_stream.dataset": "apm.internal.myservice",
 					"processor":           common.MapStr{"event": "metric", "name": "metric"},
 					"service":             common.MapStr{"name": "myservice"},
 					"span": common.MapStr{"type": spType, "subtype": spSubtype,

diff --git a/model/profile.go b/model/profile.go
@@ -128,7 +128,8 @@ func (pp PprofProfile) Transform(ctx context.Context, cfg *transform.Config) []b
 		}
 		if cfg.DataStreams {
 			event.Fields[datastreams.TypeField] = datastreams.MetricsType
-			event.Fields[datastreams.DatasetField] = ProfilesDataset
+			dataset := fmt.Sprintf("%s.%s", ProfilesDataset, datastreams.NormalizeServiceName(pp.Metadata.Service.Name))
+			event.Fields[datastreams.DatasetField] = dataset
 		}
 		var profileLabels common.MapStr
 		if len(sample.Label) > 0 {

diff --git a/model/profile_test.go b/model/profile_test.go
@@ -98,7 +98,7 @@ func TestPprofProfileTransform(t *testing.T) {
 		Timestamp: timestamp,
 		Fields: common.MapStr{
 			"data_stream.type":    "metrics",
-			"data_stream.dataset": "apm.profiling",
+			"data_stream.dataset": "apm.profiling.myservice",
 			"processor":           common.MapStr{"event": "profile", "name": "profile"},
 			"service": common.MapStr{
 				"name":        "myService",

diff --git a/model/span.go b/model/span.go
@@ -19,6 +19,7 @@ package model
 
 import (
 	"context"
+	"fmt"
 	"net"
 	"time"
 
@@ -198,7 +199,8 @@ func (e *Span) Transform(ctx context.Context, cfg *transform.Config) []beat.Even
 	if cfg.DataStreams {
 		// Spans are stored in a "traces" data stream along with transactions.
 		fields[datastreams.TypeField] = datastreams.TracesType
-		fields[datastreams.DatasetField] = TracesDataset
+		dataset := fmt.Sprintf("%s.%s", TracesDataset, datastreams.NormalizeServiceName(e.Metadata.Service.Name))
+		fields[datastreams.DatasetField] = dataset
 	}
 
 	// first set the generic metadata

diff --git a/model/span_test.go b/model/span_test.go
@@ -59,7 +59,7 @@ func TestSpanTransform(t *testing.T) {
 			Span: Span{Timestamp: timestamp, Metadata: metadata},
 			Output: common.MapStr{
 				"data_stream.type":    "traces",
-				"data_stream.dataset": "apm",
+				"data_stream.dataset": "apm.myservice",
 				"processor":           common.MapStr{"event": "span", "name": "transaction"},
 				"service":             common.MapStr{"name": serviceName, "environment": env, "version": serviceVersion},
 				"span": common.MapStr{
@@ -77,7 +77,7 @@ func TestSpanTransform(t *testing.T) {
 			Span: Span{Timestamp: timestamp, Metadata: metadata, Outcome: "success"},
 			Output: common.MapStr{
 				"data_stream.type":    "traces",
-				"data_stream.dataset": "apm",
+				"data_stream.dataset": "apm.myservice",
 				"processor":           common.MapStr{"event": "span", "name": "transaction"},
 				"service":             common.MapStr{"name": serviceName, "environment": env, "version": serviceVersion},
 				"span": common.MapStr{
@@ -126,7 +126,7 @@ func TestSpanTransform(t *testing.T) {
 			},
 			Output: common.MapStr{
 				"data_stream.type":    "traces",
-				"data_stream.dataset": "apm",
+				"data_stream.dataset": "apm.myservice",
 				"span": common.MapStr{
 					"id":       hexID,
 					"duration": common.MapStr{"us": 1200},

diff --git a/model/transaction.go b/model/transaction.go
@@ -19,6 +19,7 @@ package model
 
 import (
 	"context"
+	"fmt"
 	"time"
 
 	"github.com/elastic/beats/v7/libbeat/beat"
@@ -121,7 +122,8 @@ func (e *Transaction) Transform(_ context.Context, cfg *transform.Config) []beat
 	if cfg.DataStreams {
 		// Transactions are stored in a "traces" data stream along with spans.
 		fields[datastreams.TypeField] = datastreams.TracesType
-		fields[datastreams.DatasetField] = TracesDataset
+		dataset := fmt.Sprintf("%s.%s", TracesDataset, datastreams.NormalizeServiceName(e.Metadata.Service.Name))
+		fields[datastreams.DatasetField] = dataset
 	}
 
 	// first set generic metadata (order is relevant)

diff --git a/model/transaction_test.go b/model/transaction_test.go
@@ -182,7 +182,7 @@ func TestEventsTransformWithMetadata(t *testing.T) {
 	require.Len(t, events, 1)
 	assert.Equal(t, events[0].Fields, common.MapStr{
 		"data_stream.type":    "traces",
-		"data_stream.dataset": "apm",
+		"data_stream.dataset": "apm." + serviceName,
 		"user":                common.MapStr{"id": "123", "name": "jane"},
 		"client":              common.MapStr{"ip": ip},
 		"source":              common.MapStr{"ip": ip},

diff --git a/processor/otel/test_approved/jaeger_sampling_rate.approved.json b/processor/otel/test_approved/jaeger_sampling_rate.approved.json
@@ -6,7 +6,7 @@
                 "name": "Jaeger",
                 "version": "unknown"
             },
-            "data_stream.dataset": "apm",
+            "data_stream.dataset": "apm.unknown",
             "data_stream.type": "traces",
             "event": {
                 "outcome": "unknown"
@@ -46,7 +46,7 @@
                 "name": "Jaeger",
                 "version": "unknown"
             },
-            "data_stream.dataset": "apm",
+            "data_stream.dataset": "apm.unknown",
             "data_stream.type": "traces",
             "event": {
                 "outcome": "unknown"
@@ -90,7 +90,7 @@
                 "name": "Jaeger",
                 "version": "unknown"
             },
-            "data_stream.dataset": "apm",
+            "data_stream.dataset": "apm.unknown",
             "data_stream.type": "traces",
             "event": {
                 "outcome": "unknown"

diff --git a/processor/otel/test_approved/metadata_jaeger-no-language.approved.json b/processor/otel/test_approved/metadata_jaeger-no-language.approved.json
@@ -6,7 +6,7 @@
                 "name": "Jaeger",
                 "version": "3.4.12"
             },
-            "data_stream.dataset": "apm",
+            "data_stream.dataset": "apm.unknown",
             "data_stream.type": "traces",
             "event": {
                 "outcome": "unknown"

diff --git a/processor/otel/test_approved/metadata_jaeger-version.approved.json b/processor/otel/test_approved/metadata_jaeger-version.approved.json
@@ -6,7 +6,7 @@
                 "name": "Jaeger/PHP",
                 "version": "3.4.12"
             },
-            "data_stream.dataset": "apm",
+            "data_stream.dataset": "apm.unknown",
             "data_stream.type": "traces",
             "event": {
                 "outcome": "unknown"

diff --git a/processor/otel/test_approved/metadata_jaeger.approved.json b/processor/otel/test_approved/metadata_jaeger.approved.json
@@ -7,7 +7,7 @@
                 "name": "Jaeger/C++",
                 "version": "3.2.1"
             },
-            "data_stream.dataset": "apm",
+            "data_stream.dataset": "apm.foo",
             "data_stream.type": "traces",
             "event": {
                 "outcome": "unknown"

diff --git a/processor/otel/test_approved/span_jaeger_custom.approved.json b/processor/otel/test_approved/span_jaeger_custom.approved.json
@@ -6,7 +6,7 @@
                 "name": "Jaeger",
                 "version": "unknown"
             },
-            "data_stream.dataset": "apm",
+            "data_stream.dataset": "apm.unknown",
             "data_stream.type": "traces",
             "event": {
                 "outcome": "unknown"

diff --git a/processor/otel/test_approved/span_jaeger_db.approved.json b/processor/otel/test_approved/span_jaeger_db.approved.json
@@ -6,7 +6,7 @@
                 "name": "Jaeger",
                 "version": "unknown"
             },
-            "data_stream.dataset": "apm",
+            "data_stream.dataset": "apm.unknown",
             "data_stream.type": "traces",
             "destination": {
                 "address": "db",

diff --git a/processor/otel/test_approved/span_jaeger_http.approved.json b/processor/otel/test_approved/span_jaeger_http.approved.json
@@ -6,7 +6,7 @@
                 "name": "Jaeger",
                 "version": "unknown"
             },
-            "data_stream.dataset": "apm",
+            "data_stream.dataset": "apm.unknown",
             "data_stream.type": "traces",
             "destination": {
                 "address": "foo.bar.com",
@@ -80,7 +80,7 @@
                 "name": "Jaeger",
                 "version": "unknown"
             },
-            "data_stream.dataset": "apm.error",
+            "data_stream.dataset": "apm.error.unknown",
             "data_stream.type": "logs",
             "error": {
                 "exception": [
@@ -141,7 +141,7 @@
                 "name": "Jaeger",
                 "version": "unknown"
             },
-            "data_stream.dataset": "apm.error",
+            "data_stream.dataset": "apm.error.unknown",
             "data_stream.type": "logs",
             "error": {
                 "grouping_key": "23b7ac1bdf1ca957f9f581cfadee467c",
@@ -197,7 +197,7 @@
                 "name": "Jaeger",
                 "version": "unknown"
             },
-            "data_stream.dataset": "apm.error",
+            "data_stream.dataset": "apm.error.unknown",
             "data_stream.type": "logs",
             "error": {
                 "exception": [
@@ -255,7 +255,7 @@
                 "name": "Jaeger",
                 "version": "unknown"
             },
-            "data_stream.dataset": "apm.error",
+            "data_stream.dataset": "apm.error.unknown",
             "data_stream.type": "logs",
             "error": {
                 "exception": [
@@ -313,7 +313,7 @@
                 "name": "Jaeger",
                 "version": "unknown"
             },
-            "data_stream.dataset": "apm.error",
+            "data_stream.dataset": "apm.error.unknown",
             "data_stream.type": "logs",
             "error": {
                 "exception": [
@@ -371,7 +371,7 @@
                 "name": "Jaeger",
                 "version": "unknown"
             },
-            "data_stream.dataset": "apm.error",
+            "data_stream.dataset": "apm.error.unknown",
             "data_stream.type": "logs",
             "error": {
                 "grouping_key": "c9221918248f05433f6b81c46a666aee",

diff --git a/processor/otel/test_approved/span_jaeger_http_status_code.approved.json b/processor/otel/test_approved/span_jaeger_http_status_code.approved.json
@@ -6,7 +6,7 @@
                 "name": "Jaeger",
                 "version": "unknown"
             },
-            "data_stream.dataset": "apm",
+            "data_stream.dataset": "apm.unknown",
             "data_stream.type": "traces",
             "destination": {
                 "address": "foo.bar.com",

diff --git a/processor/otel/test_approved/span_jaeger_https_default_port.approved.json b/processor/otel/test_approved/span_jaeger_https_default_port.approved.json
@@ -6,7 +6,7 @@
                 "name": "Jaeger",
                 "version": "unknown"
             },
-            "data_stream.dataset": "apm",
+            "data_stream.dataset": "apm.unknown",
             "data_stream.type": "traces",
             "destination": {
                 "address": "foo.bar.com",

diff --git a/processor/otel/test_approved/span_jaeger_messaging.approved.json b/processor/otel/test_approved/span_jaeger_messaging.approved.json
@@ -6,7 +6,7 @@
                 "name": "Jaeger",
                 "version": "unknown"
             },
-            "data_stream.dataset": "apm",
+            "data_stream.dataset": "apm.unknown",
             "data_stream.type": "traces",
             "destination": {
                 "address": "mq",

diff --git a/processor/otel/test_approved/span_jaeger_subtype_component.approved.json b/processor/otel/test_approved/span_jaeger_subtype_component.approved.json
@@ -6,7 +6,7 @@
                 "name": "Jaeger",
                 "version": "unknown"
             },
-            "data_stream.dataset": "apm",
+            "data_stream.dataset": "apm.unknown",
             "data_stream.type": "traces",
             "event": {
                 "outcome": "unknown"

diff --git a/processor/otel/test_approved/transaction_jaeger_custom.approved.json b/processor/otel/test_approved/transaction_jaeger_custom.approved.json
@@ -6,7 +6,7 @@
                 "name": "Jaeger",
                 "version": "unknown"
             },
-            "data_stream.dataset": "apm",
+            "data_stream.dataset": "apm.unknown",
             "data_stream.type": "traces",
             "event": {
                 "outcome": "unknown"