improve util helper functions

elastic · Sep 20, 2024 · a6450a9 · a6450a9
1 parent 6a3dbe9
commit a6450a9
Show file tree

Hide file tree

Showing 7 changed files with 53 additions and 47 deletions.
diff --git a/input/otlp/logs.go b/input/otlp/logs.go
@@ -48,11 +48,6 @@ import (
 	"github.com/elastic/apm-data/model/modelpb"
 )
 
-const (
-	MaxDataStreamBytes        = 100
-	DisallowedDataStreamRunes = "-\\/*?\"<>| ,#"
-)
-
 // ConsumeLogsResult contains the number of rejected log records and error message for partial success response.
 type ConsumeLogsResult struct {
 	ErrorMessage       string
@@ -244,21 +239,3 @@ func (c *Consumer) convertLogRecord(
 
 	return event
 }
-
-// Sanitize the datastream fields (dataset, namespace) to apply restrictions
-// as outlined in https://www.elastic.co/guide/en/ecs/current/ecs-data_stream.html
-func sanitizeDataStreamField(field string) string {
-	field = strings.ToLower(field)
-	field = strings.Map(replaceReservedRune, field)
-	if len(field) > MaxDataStreamBytes {
-		return field[:MaxDataStreamBytes]
-	}
-	return field
-}
-
-func replaceReservedRune(r rune) rune {
-	if strings.ContainsRune(DisallowedDataStreamRunes, r) {
-		return '_'
-	}
-	return r
-}
diff --git a/input/otlp/logs_test.go b/input/otlp/logs_test.go
@@ -36,9 +36,7 @@ package otlp_test
 
 import (
 	"context"
-	"crypto/rand"
 	"fmt"
-	"math/big"
 	"testing"
 	"time"
 
@@ -579,8 +577,8 @@ func TestConsumerConsumeLogsDataStream(t *testing.T) {
 		{
 			resourceDataStreamDataset:   otlp.DisallowedDataStreamRunes + randomString,
 			resourceDataStreamNamespace: otlp.DisallowedDataStreamRunes + randomString,
-			expectedDataStreamDataset:   "____________" + randomString[:maxLen],
-			expectedDataStreamNamespace: "____________" + randomString[:maxLen],
+			expectedDataStreamDataset:   revertedString(otlp.DisallowedDataStreamRunes) + randomString[:maxLen],
+			expectedDataStreamNamespace: revertedString(otlp.DisallowedDataStreamRunes) + randomString[:maxLen],
 		},
 	} {
 		tcName := fmt.Sprintf("%s,%s", tc.expectedDataStreamDataset, tc.expectedDataStreamNamespace)
@@ -846,17 +844,3 @@ func newLogRecord(body interface{}) plog.LogRecord {
 	}
 	return otelLogRecord
 }
-
-const charset = "abcdefghijklmnopqrstuvwxyz0123456789"
-
-func generateRandomString(length int) (string, error) {
-	result := make([]byte, length)
-	for i := range result {
-		index, err := rand.Int(rand.Reader, big.NewInt(int64(len(charset))))
-		if err != nil {
-			return "", err
-		}
-		result[i] = charset[index.Int64()]
-	}
-	return string(result), nil
-}
diff --git a/input/otlp/metadata.go b/input/otlp/metadata.go
@@ -30,7 +30,9 @@ import (
 )
 
 const (
-	AgentNameJaeger = "Jaeger"
+	AgentNameJaeger           = "Jaeger"
+	MaxDataStreamBytes        = 100
+	DisallowedDataStreamRunes = "-\\/*?\"<>| ,#:"
 )
 
 var (
@@ -545,3 +547,21 @@ func setLabel(key string, event *modelpb.APMEvent, v pcommon.Value) {
 		}
 	}
 }
+
+// Sanitize the datastream fields (dataset, namespace) to apply restrictions
+// as outlined in https://www.elastic.co/guide/en/ecs/current/ecs-data_stream.html
+func sanitizeDataStreamField(field string) string {
+	field = strings.ToLower(field)
+	field = strings.Map(replaceReservedRune, field)
+	if len(field) > MaxDataStreamBytes {
+		return field[:MaxDataStreamBytes]
+	}
+	return field
+}
+
+func replaceReservedRune(r rune) rune {
+	if strings.ContainsRune(DisallowedDataStreamRunes, r) {
+		return '_'
+	}
+	return r
+}
diff --git a/input/otlp/metadata_test.go b/input/otlp/metadata_test.go
@@ -18,6 +18,8 @@
 package otlp_test
 
 import (
+	"crypto/rand"
+	"math/big"
 	"testing"
 
 	"github.com/google/go-cmp/cmp"
@@ -395,3 +397,26 @@ func transformResourceMetadata(t *testing.T, resourceAttrs map[string]interface{
 	(*events)[0].Timestamp = 0
 	return (*events)[0]
 }
+
+const charset = "abcdefghijklmnopqrstuvwxyz0123456789"
+
+func generateRandomString(length int) (string, error) {
+	result := make([]byte, length)
+	for i := range result {
+		index, err := rand.Int(rand.Reader, big.NewInt(int64(len(charset))))
+		if err != nil {
+			return "", err
+		}
+		result[i] = charset[index.Int64()]
+	}
+	return string(result), nil
+}
+
+// Creates a string consistening of _ that equals the length of otlp.DisallowedDataStreamRunes
+func revertedString(s string) string {
+	var res []rune
+	for i := 0; i < len(s); i++ {
+		res = append(res, '_')
+	}
+	return string(res)
+}
diff --git a/input/otlp/metrics.go b/input/otlp/metrics.go
@@ -180,7 +180,7 @@ func (c *Consumer) handleScopeMetrics(
 					if event.DataStream == nil {
 						event.DataStream = &modelpb.DataStream{}
 					}
-					event.DataStream.Dataset = v.Str()
+					event.DataStream.Dataset = sanitizeDataStreamField(v.Str())
 				case attributeDataStreamNamespace:
 					if event.DataStream == nil {
 						event.DataStream = &modelpb.DataStream{}

diff --git a/input/otlp/metrics_test.go b/input/otlp/metrics_test.go
@@ -774,8 +774,8 @@ func TestConsumeMetricsDataStream(t *testing.T) {
 		{
 			resourceDataStreamDataset:   otlp.DisallowedDataStreamRunes + randomString,
 			resourceDataStreamNamespace: otlp.DisallowedDataStreamRunes + randomString,
-			expectedDataStreamDataset:   "____________" + randomString[:maxLen],
-			expectedDataStreamNamespace: "____________" + randomString[:maxLen],
+			expectedDataStreamDataset:   revertedString(otlp.DisallowedDataStreamRunes) + randomString[:maxLen],
+			expectedDataStreamNamespace: revertedString(otlp.DisallowedDataStreamRunes) + randomString[:maxLen],
 		},
 	} {
 		tcName := fmt.Sprintf("%s,%s", tc.expectedDataStreamDataset, tc.expectedDataStreamNamespace)

diff --git a/input/otlp/traces_test.go b/input/otlp/traces_test.go
@@ -918,8 +918,8 @@ func TestSpanDataStream(t *testing.T) {
 		{
 			resourceDataStreamDataset:   otlp.DisallowedDataStreamRunes + randomString,
 			resourceDataStreamNamespace: otlp.DisallowedDataStreamRunes + randomString,
-			expectedDataStreamDataset:   "____________" + randomString[:maxLen],
-			expectedDataStreamNamespace: "____________" + randomString[:maxLen],
+			expectedDataStreamDataset:   revertedString(otlp.DisallowedDataStreamRunes) + randomString[:maxLen],
+			expectedDataStreamNamespace: revertedString(otlp.DisallowedDataStreamRunes) + randomString[:maxLen],
 		},
 	} {
 		for _, isTxn := range []bool{false, true} {