Override RUM data stream fields

carsonip · Jan 24, 2024 · 14781c7 · 14781c7
1 parent 9c48095
commit 14781c7
Show file tree

Hide file tree

Showing 2 changed files with 11 additions and 9 deletions.
diff --git a/model/modelprocessor/datastream.go b/model/modelprocessor/datastream.go
@@ -84,13 +84,13 @@ func (s *SetDataStream) setDataStream(event *modelpb.APMEvent) {
 		event.DataStream.Dataset = errorsDataset
 	case modelpb.LogEventType:
 		event.DataStream.Type = logsType
-		if event.DataStream.Dataset == "" {
+		if event.DataStream.Dataset == "" || isRUMAgentName(event.GetAgent().GetName()) {
 			// Only set dataset if it is not already set in the input event
 			event.DataStream.Dataset = getAppLogsDataset(event)
 		}
 	case modelpb.MetricEventType:
 		event.DataStream.Type = metricsType
-		if event.DataStream.Dataset == "" {
+		if event.DataStream.Dataset == "" || isRUMAgentName(event.GetAgent().GetName()) {
 			// Only set dataset if it is not already set in the input event
 			event.DataStream.Dataset = metricsetDataset(event)
 		}

diff --git a/model/modelprocessor/datastream_test.go b/model/modelprocessor/datastream_test.go
@@ -106,7 +106,7 @@ func TestSetDataStream(t *testing.T) {
 			Service:    &modelpb.Service{Name: "service-name"},
 			DataStream: &modelpb.DataStream{Dataset: "dataset", Namespace: "namespace"},
 		},
-		output: &modelpb.DataStream{Type: "logs", Dataset: "dataset", Namespace: "namespace"},
+		output: &modelpb.DataStream{Type: "logs", Dataset: "apm.app.service_name", Namespace: "custom"},
 	}, {
 		input: &modelpb.APMEvent{
 			Agent:       &modelpb.Agent{Name: "rum-js"},
@@ -150,7 +150,7 @@ func TestSetDataStream(t *testing.T) {
 			},
 			DataStream: &modelpb.DataStream{Dataset: "dataset", Namespace: "namespace"},
 		},
-		output: &modelpb.DataStream{Type: "metrics", Dataset: "dataset", Namespace: "namespace"},
+		output: &modelpb.DataStream{Type: "metrics", Dataset: "apm.app.service_name", Namespace: "custom"},
 	}, {
 		input: &modelpb.APMEvent{
 			Service:     &modelpb.Service{Name: "service-name"},
@@ -193,11 +193,13 @@ func TestSetDataStream(t *testing.T) {
 	}}
 
 	for _, test := range tests {
-		batch := modelpb.Batch{test.input}
-		processor := modelprocessor.SetDataStream{Namespace: "custom"}
-		err := processor.ProcessBatch(context.Background(), &batch)
-		assert.NoError(t, err)
-		assert.Equal(t, test.output, batch[0].DataStream)
+		t.Run("", func(t *testing.T) {
+			batch := modelpb.Batch{test.input}
+			processor := modelprocessor.SetDataStream{Namespace: "custom"}
+			err := processor.ProcessBatch(context.Background(), &batch)
+			assert.NoError(t, err)
+			assert.Equal(t, test.output, batch[0].DataStream)
+		})
 	}
 
 }