[receiver/cloudflare] Support map fields

.chloggen/feat-cloudflare-map.yaml

@@ -0,0 +1,27 @@
+# Use this changelog template to create an entry for release notes.
+
+# One of 'breaking', 'deprecation', 'new_component', 'enhancement', 'bug_fix'
+change_type: "enhancement"
+
+# The name of the component, or a single word describing the area of concern, (e.g. filelogreceiver)
+component: "receiver/cloudflare"
+
+# A brief description of the change.  Surround your text with quotes ("") if it needs to start with a backtick (`).
+note: Enable the receiver to consume fields from Cloudflare containing a map
+
+# Mandatory: One or more tracking issues related to the change. You can use the PR number here if no issue exists.
+issues: [40318]
+
+# (Optional) One or more lines of additional information to render under the primary note.
+# These lines will be padded with 2 spaces and then inserted directly into the document.
+# Use pipe (|) for multiline entries.
+subtext:
+
+# If your change doesn't affect end users or the exported elements of any package,
+# you should instead start your pull request title with [chore] or use the "Skip Changelog" label.
+# Optional: The change log or logs in which this entry should be included.
+# e.g. '[user]' or '[user, api]'
+# Include 'user' if the change is relevant to end users.
+# Include 'api' if there is a change to a library API.
+# Default: '[user]'
+change_logs: [user]

receiver/cloudflarereceiver/README.md

@@ -50,6 +50,8 @@ If the receiver will be handling TLS termination:
 - `attributes`
   - This parameter allows the receiver to be configured to set log record attributes based on fields found in the log message. The fields are not removed from the log message when set in this way. Only string, boolean, integer or float fields can be mapped using this parameter.
   - When the `attributes` configuration is empty, the receiver will automatically ingest all fields from the log messages as attributes, using the original field names as attribute names.
+- `separator` (default: `.`)
+  - The separator used to join nested fields in the log message when setting attributes. For example, if the log message contains a field `"RequestHeaders": { "Content-Type": "application/json" }`, and the `separator` is set to `.`, the attribute will be set as `RequestHeaders.Content_Type`. If the separator is set to `_`, it will be set as `RequestHeaders_Content_Type`.
 
 
 ### Example:

receiver/cloudflarereceiver/config.go

@@ -26,6 +26,7 @@ type LogsConfig struct {
 	TLS            *configtls.ServerConfig `mapstructure:"tls"`
 	Attributes     map[string]string       `mapstructure:"attributes"`
 	TimestampField string                  `mapstructure:"timestamp_field"`
+	Separator      string                  `mapstructure:"separator"`
 
 	// prevent unkeyed literal initialization
 	_ struct{}
@@ -37,6 +38,7 @@ var (
 	errNoKey      = errors.New("tls was configured, but no key file was specified")
 
 	defaultTimestampField = "EdgeStartTimestamp"
+	defaultSeparator      = "."
 )
 
 func (c *Config) Validate() error {

receiver/cloudflarereceiver/config_test.go

@@ -123,6 +123,7 @@ func TestLoadConfig(t *testing.T) {
 					},
 					Secret:         "1234567890abcdef1234567890abcdef",
 					TimestampField: "EdgeStartTimestamp",
+					Separator:      ".",
 					Attributes: map[string]string{
 						"ClientIP":         "http_request.client_ip",
 						"ClientRequestURI": "http_request.uri",

receiver/cloudflarereceiver/factory.go

@@ -36,6 +36,7 @@ func createDefaultConfig() component.Config {
 	return &Config{
 		Logs: LogsConfig{
 			TimestampField: defaultTimestampField,
+			Separator:      defaultSeparator,
 		},
 	}
 }

receiver/cloudflarereceiver/logs.go

@@ -11,9 +11,11 @@ import (
 	"errors"
 	"fmt"
 	"io"
+	"maps"
 	"net"
 	"net/http"
 	"strconv"
+	"strings"
 	"sync"
 	"time"
 
@@ -310,6 +312,28 @@ func (l *logsReceiver) processLogs(now pcommon.Timestamp, logs []map[string]any)
 					attrs.PutDouble(attrName, v)
 				case bool:
 					attrs.PutBool(attrName, v)
+				case map[string]any:
+					// Flatten the map and add each field with a prefixed key
+					flattened := flattenMap(v, attrName+l.cfg.Separator, l.cfg.Separator)
+					for k, val := range flattened {
+						switch v := val.(type) {
+						case string:
+							attrs.PutStr(k, v)
+						case int:
+							attrs.PutInt(k, int64(v))
+						case int64:
+							attrs.PutInt(k, v)
+						case float64:
+							attrs.PutDouble(k, v)
+						case bool:
+							attrs.PutBool(k, v)
+						default:
+							l.logger.Warn("unable to translate flattened field to attribute, unsupported type",
+								zap.String("field", k),
+								zap.Any("value", v),
+								zap.String("type", fmt.Sprintf("%T", v)))
+						}
+					}
 				default:
 					l.logger.Warn("unable to translate field to attribute, unsupported type",
 						zap.String("field", field),
@@ -343,3 +367,23 @@ func severityFromStatusCode(statusCode int64) plog.SeverityNumber {
 		return plog.SeverityNumberUnspecified
 	}
 }
+
+// flattenMap recursively flattens a map[string]interface{} into a single level map
+// with keys joined by dot.
+func flattenMap(input map[string]any, prefix string, separator string) map[string]any {
+	result := make(map[string]any)
+	for k, v := range input {
+		// Replace hyphens with underscores in the key. Content-Type becomes Content_Type
+		k = strings.ReplaceAll(k, "-", "_")
+		newKey := prefix + k
+		switch val := v.(type) {
+		case map[string]any:
+			// Recursively flatten nested maps
+			nested := flattenMap(val, newKey+separator, separator)
+			maps.Copy(result, nested)
+		default:
+			result[newKey] = v
+		}
+	}
+	return result
+}

receiver/cloudflarereceiver/logs_test.go

@@ -363,14 +363,14 @@ func TestEmptyAttributes(t *testing.T) {
 	}{
 		{
 			name: "Empty Attributes Map",
-			payload: `{ "ClientIP": "89.163.253.200", "ClientRequestHost": "www.theburritobot0.com", "ClientRequestMethod": "GET", "ClientRequestURI": "/static/img/testimonial-hipster.png", "EdgeEndTimestamp": "2023-03-03T05:30:05Z", "EdgeResponseBytes": "69045", "EdgeResponseStatus": "200", "EdgeStartTimestamp": "2023-03-03T05:29:05Z", "RayID": "3a6050bcbe121a87" }
-{ "ClientIP" : "89.163.253.201", "ClientRequestHost": "www.theburritobot1.com", "ClientRequestMethod": "GET", "ClientRequestURI": "/static/img/testimonial-hipster.png", "EdgeEndTimestamp": "2023-03-03T05:30:05Z", "EdgeResponseBytes": "69045", "EdgeResponseStatus": "200", "EdgeStartTimestamp": "2023-03-03T05:29:05Z", "RayID": "3a6050bcbe121a87" }`,
+			payload: `{ "ClientIP": "89.163.253.200", "ClientRequestHost": "www.theburritobot0.com", "ClientRequestMethod": "GET", "ClientRequestURI": "/static/img/testimonial-hipster.png", "EdgeEndTimestamp": "2023-03-03T05:30:05Z", "EdgeResponseBytes": "69045", "EdgeResponseStatus": "200", "EdgeStartTimestamp": "2023-03-03T05:29:05Z", "RayID": "3a6050bcbe121a87", "RequestHeaders": { "Content-Type": "application/json" } }
+{ "ClientIP" : "89.163.253.201", "ClientRequestHost": "www.theburritobot1.com", "ClientRequestMethod": "GET", "ClientRequestURI": "/static/img/testimonial-hipster.png", "EdgeEndTimestamp": "2023-03-03T05:30:05Z", "EdgeResponseBytes": "69045", "EdgeResponseStatus": "200", "EdgeStartTimestamp": "2023-03-03T05:29:05Z", "RayID": "3a6050bcbe121a87", "RequestHeaders": { "Content-Type": "application/json" } }`,
 			attributes: map[string]string{},
 		},
 		{
 			name: "nil Attributes",
-			payload: `{ "ClientIP": "89.163.253.200", "ClientRequestHost": "www.theburritobot0.com", "ClientRequestMethod": "GET", "ClientRequestURI": "/static/img/testimonial-hipster.png", "EdgeEndTimestamp": "2023-03-03T05:30:05Z", "EdgeResponseBytes": "69045", "EdgeResponseStatus": "200", "EdgeStartTimestamp": "2023-03-03T05:29:05Z", "RayID": "3a6050bcbe121a87" }
-{ "ClientIP" : "89.163.253.201", "ClientRequestHost": "www.theburritobot1.com", "ClientRequestMethod": "GET", "ClientRequestURI": "/static/img/testimonial-hipster.png", "EdgeEndTimestamp": "2023-03-03T05:30:05Z", "EdgeResponseBytes": "69045", "EdgeResponseStatus": "200", "EdgeStartTimestamp": "2023-03-03T05:29:05Z", "RayID": "3a6050bcbe121a87" }`,
+			payload: `{ "ClientIP": "89.163.253.200", "ClientRequestHost": "www.theburritobot0.com", "ClientRequestMethod": "GET", "ClientRequestURI": "/static/img/testimonial-hipster.png", "EdgeEndTimestamp": "2023-03-03T05:30:05Z", "EdgeResponseBytes": "69045", "EdgeResponseStatus": "200", "EdgeStartTimestamp": "2023-03-03T05:29:05Z", "RayID": "3a6050bcbe121a87", "RequestHeaders": { "Content-Type": "application/json" } }
+{ "ClientIP" : "89.163.253.201", "ClientRequestHost": "www.theburritobot1.com", "ClientRequestMethod": "GET", "ClientRequestURI": "/static/img/testimonial-hipster.png", "EdgeEndTimestamp": "2023-03-03T05:30:05Z", "EdgeResponseBytes": "69045", "EdgeResponseStatus": "200", "EdgeStartTimestamp": "2023-03-03T05:29:05Z", "RayID": "3a6050bcbe121a87", "RequestHeaders": { "Content-Type": "application/json" } }`,
 			attributes: nil,
 		},
 	}
@@ -381,6 +381,95 @@ func TestEmptyAttributes(t *testing.T) {
 		sl := rl.ScopeLogs().AppendEmpty()
 		sl.Scope().SetName("github.com/open-telemetry/opentelemetry-collector-contrib/receiver/cloudflarereceiver")
 
+		for idx, line := range strings.Split(payload, "\n") {
+			lr := sl.LogRecords().AppendEmpty()
+
+			require.NoError(t, lr.Attributes().FromRaw(map[string]any{
+				"ClientIP":                    fmt.Sprintf("89.163.253.%d", 200+idx),
+				"ClientRequestHost":           fmt.Sprintf("www.theburritobot%d.com", idx),
+				"ClientRequestMethod":         "GET",
+				"ClientRequestURI":            "/static/img/testimonial-hipster.png",
+				"EdgeEndTimestamp":            "2023-03-03T05:30:05Z",
+				"EdgeResponseBytes":           "69045",
+				"EdgeResponseStatus":          "200",
+				"EdgeStartTimestamp":          "2023-03-03T05:29:05Z",
+				"RayID":                       "3a6050bcbe121a87",
+				"RequestHeaders.Content_Type": "application/json",
+			}))
+
+			lr.SetObservedTimestamp(pcommon.NewTimestampFromTime(now))
+			ts, err := time.Parse(time.RFC3339, "2023-03-03T05:29:05Z")
+			require.NoError(t, err)
+			lr.SetTimestamp(pcommon.NewTimestampFromTime(ts))
+			lr.SetSeverityNumber(plog.SeverityNumberInfo)
+			lr.SetSeverityText(plog.SeverityNumberInfo.String())
+
+			var log map[string]any
+			err = json.Unmarshal([]byte(line), &log)
+			require.NoError(t, err)
+
+			payloadToExpectedBody(t, line, lr)
+		}
+		fmt.Println(logs)
+		return logs
+	}
+
+	for _, tc := range testCases {
+		t.Run(tc.name, func(t *testing.T) {
+			recv := newReceiver(t, &Config{
+				Logs: LogsConfig{
+					Endpoint:       "localhost:0",
+					TLS:            &configtls.ServerConfig{},
+					TimestampField: "EdgeStartTimestamp",
+					Attributes:     tc.attributes,
+					Separator:      ".",
+				},
+			},
+				&consumertest.LogsSink{},
+			)
+			var logs plog.Logs
+			rawLogs, err := parsePayload([]byte(tc.payload))
+			if err == nil {
+				logs = recv.processLogs(pcommon.NewTimestampFromTime(time.Now()), rawLogs)
+			}
+			require.NoError(t, err)
+			require.NotNil(t, logs)
+			require.NoError(t, plogtest.CompareLogs(expectedLogs(t, tc.payload), logs, plogtest.IgnoreObservedTimestamp()))
+		})
+	}
+}
+
+func TestAttributesWithSeparator(t *testing.T) {
+	now := time.Time{}
+
+	testCases := []struct {
+		name       string
+		payload    string
+		attributes map[string]string
+		separator  string
+	}{
+		{
+			name: "Empty Attributes Map",
+			payload: `{ "ClientIP": "89.163.253.200", "ClientRequestHost": "www.theburritobot0.com", "ClientRequestMethod": "GET", "ClientRequestURI": "/static/img/testimonial-hipster.png", "EdgeEndTimestamp": "2023-03-03T05:30:05Z", "EdgeResponseBytes": "69045", "EdgeResponseStatus": "200", "EdgeStartTimestamp": "2023-03-03T05:29:05Z", "RayID": "3a6050bcbe121a87", "RequestHeaders": { "Content-Type": "application/json" } }
+{ "ClientIP" : "89.163.253.201", "ClientRequestHost": "www.theburritobot1.com", "ClientRequestMethod": "GET", "ClientRequestURI": "/static/img/testimonial-hipster.png", "EdgeEndTimestamp": "2023-03-03T05:30:05Z", "EdgeResponseBytes": "69045", "EdgeResponseStatus": "200", "EdgeStartTimestamp": "2023-03-03T05:29:05Z", "RayID": "3a6050bcbe121a87", "RequestHeaders": { "Content-Type": "application/json" } }`,
+			attributes: map[string]string{},
+			separator:  ".",
+		},
+		{
+			name: "nil Attributes",
+			payload: `{ "ClientIP": "89.163.253.200", "ClientRequestHost": "www.theburritobot0.com", "ClientRequestMethod": "GET", "ClientRequestURI": "/static/img/testimonial-hipster.png", "EdgeEndTimestamp": "2023-03-03T05:30:05Z", "EdgeResponseBytes": "69045", "EdgeResponseStatus": "200", "EdgeStartTimestamp": "2023-03-03T05:29:05Z", "RayID": "3a6050bcbe121a87", "RequestHeaders": { "Content-Type": "application/json" } }
+{ "ClientIP" : "89.163.253.201", "ClientRequestHost": "www.theburritobot1.com", "ClientRequestMethod": "GET", "ClientRequestURI": "/static/img/testimonial-hipster.png", "EdgeEndTimestamp": "2023-03-03T05:30:05Z", "EdgeResponseBytes": "69045", "EdgeResponseStatus": "200", "EdgeStartTimestamp": "2023-03-03T05:29:05Z", "RayID": "3a6050bcbe121a87", "RequestHeaders": { "Content-Type": "application/json" } }`,
+			attributes: nil,
+			separator:  "_test_",
+		},
+	}
+
+	expectedLogs := func(t *testing.T, payload string, separator string) plog.Logs {
+		logs := plog.NewLogs()
+		rl := logs.ResourceLogs().AppendEmpty()
+		sl := rl.ScopeLogs().AppendEmpty()
+		sl.Scope().SetName("github.com/open-telemetry/opentelemetry-collector-contrib/receiver/cloudflarereceiver")
+
 		for idx, line := range strings.Split(payload, "\n") {
 			lr := sl.LogRecords().AppendEmpty()
 
@@ -394,6 +483,7 @@ func TestEmptyAttributes(t *testing.T) {
 				"EdgeResponseStatus":  "200",
 				"EdgeStartTimestamp":  "2023-03-03T05:29:05Z",
 				"RayID":               "3a6050bcbe121a87",
+				fmt.Sprintf("RequestHeaders%sContent_Type", separator): "application/json",
 			}))
 
 			lr.SetObservedTimestamp(pcommon.NewTimestampFromTime(now))
@@ -409,7 +499,7 @@ func TestEmptyAttributes(t *testing.T) {
 
 			payloadToExpectedBody(t, line, lr)
 		}
-
+		fmt.Println(logs)
 		return logs
 	}
 
@@ -421,6 +511,7 @@ func TestEmptyAttributes(t *testing.T) {
 					TLS:            &configtls.ServerConfig{},
 					TimestampField: "EdgeStartTimestamp",
 					Attributes:     tc.attributes,
+					Separator:      tc.separator,
 				},
 			},
 				&consumertest.LogsSink{},
@@ -432,7 +523,7 @@ func TestEmptyAttributes(t *testing.T) {
 			}
 			require.NoError(t, err)
 			require.NotNil(t, logs)
-			require.NoError(t, plogtest.CompareLogs(expectedLogs(t, tc.payload), logs, plogtest.IgnoreObservedTimestamp()))
+			require.NoError(t, plogtest.CompareLogs(expectedLogs(t, tc.payload, tc.separator), logs, plogtest.IgnoreObservedTimestamp()))
 		})
 	}
 }