[exporter/datasetexporter]: Fix NPE exception when attribute contains…

… nil (open-telemetry#27663)

**Description:** Nil is valid value for the attributes in events, spans,
resources, and scopes. Lets do not crash when it appears.

**Link to tracking Issue:** open-telemetry#27648

**Testing:** I have added unit tests to verify the fix.

**Documentation:** <Describe the documentation added.>

.chloggen/27648-dataset-fix-npe.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: bug_fix
+
+# The name of the component, or a single word describing the area of concern, (e.g. filelogreceiver)
+component: datasetexporter
+
+# A brief description of the change.  Surround your text with quotes ("") if it needs to start with a backtick (`).
+note: Do not crash on NPE when any of the attributes contains null value.
+
+# Mandatory: One or more tracking issues related to the change. You can use the PR number here if no issue exists.
+issues: [27648]
+
+# (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: []

exporter/datasetexporter/datasetexporter.go

@@ -102,6 +102,10 @@ func updateWithPrefixedValuesArray(target map[string]interface{}, prefix string,
 
 func updateWithPrefixedValues(target map[string]interface{}, prefix string, separator string, source interface{}, depth int) {
 	st := reflect.TypeOf(source)
+	if st == nil {
+		target[prefix] = source
+		return
+	}
 	switch st.Kind() {
 	case reflect.Map:
 		updateWithPrefixedValuesMap(target, prefix, separator, source.(map[string]interface{}), depth)

exporter/datasetexporter/datasetexporter_test.go

@@ -2,3 +2,23 @@
 // SPDX-License-Identifier: Apache-2.0
 
 package datasetexporter
+
+import "go.opentelemetry.io/collector/pdata/pcommon"
+
+func fillAttributes(attr pcommon.Map) {
+	attr.PutStr("string", "string")
+	attr.PutDouble("double", 2.0)
+	attr.PutBool("bool", true)
+	attr.PutEmpty("empty")
+	attr.PutInt("int", 3)
+
+	attr.PutEmptyMap("empty_map")
+	mVal := attr.PutEmptyMap("map")
+	mVal.PutStr("map_string", "map_string")
+	mVal.PutEmpty("map_empty")
+
+	attr.PutEmptySlice("empty_slice")
+	sVal := attr.PutEmptySlice("slice")
+	sVal.AppendEmpty()
+	sVal.At(0).SetStr("slice_string")
+}

exporter/datasetexporter/logs_exporter_test.go

@@ -411,8 +411,11 @@ func TestConsumeLogsShouldSucceed(t *testing.T) {
 			RetryMaxElapsedTime:  time.Hour,
 			RetryShutdownTimeout: time.Minute,
 		},
-		LogsSettings:   newDefaultLogsSettings(),
-		TracesSettings: newDefaultTracesSettings(),
+		LogsSettings: LogsSettings{
+			ExportResourceInfo: true,
+			ExportScopeInfo:    true,
+		},
+		TracesSettings: TracesSettings{},
 		ServerHostSettings: ServerHostSettings{
 			ServerHost: testServerHost,
 		},
@@ -433,16 +436,24 @@ func TestConsumeLogsShouldSucceed(t *testing.T) {
 	// set attribute host.name in the resource attribute
 	lr4 := testdata.GenerateLogsOneLogRecord()
 	lr4.ResourceLogs().At(0).Resource().Attributes().PutStr("host.name", "serverHostFromResourceHost")
+	// set all possible values
+	lr5 := testdata.GenerateLogsOneLogRecord()
+	fillAttributes(lr5.ResourceLogs().At(0).Resource().Attributes())
+	fillAttributes(lr5.ResourceLogs().At(0).ScopeLogs().At(0).Scope().Attributes())
+	fillAttributes(lr5.ResourceLogs().At(0).ScopeLogs().At(0).LogRecords().At(0).Attributes())
 
 	ld := plog.NewLogs()
 	ld.ResourceLogs().AppendEmpty()
 	ld.ResourceLogs().AppendEmpty()
 	ld.ResourceLogs().AppendEmpty()
 	ld.ResourceLogs().AppendEmpty()
+	ld.ResourceLogs().AppendEmpty()
+
 	lr1.ResourceLogs().At(0).CopyTo(ld.ResourceLogs().At(0))
 	lr2.ResourceLogs().At(0).CopyTo(ld.ResourceLogs().At(1))
 	lr3.ResourceLogs().At(0).CopyTo(ld.ResourceLogs().At(2))
 	lr4.ResourceLogs().At(0).CopyTo(ld.ResourceLogs().At(3))
+	lr5.ResourceLogs().At(0).CopyTo(ld.ResourceLogs().At(4))
 
 	logs, err := createLogsExporter(context.Background(), createSettings, config)
 	if assert.NoError(t, err) {
@@ -467,7 +478,24 @@ func TestConsumeLogsShouldSucceed(t *testing.T) {
 				Session:     addRequest.Session,
 				SessionInfo: addRequest.SessionInfo,
 				Events: []*add_events.Event{
-					testLEventReq,
+					{
+						Thread: testLEventReq.Thread,
+						Log:    testLEventReq.Log,
+						Sev:    testLEventReq.Sev,
+						Ts:     testLEventReq.Ts,
+						Attrs: map[string]interface{}{
+							add_events.AttrOrigServerHost: testServerHost,
+							"app":                         "server",
+							"instance_num":                float64(1),
+							"dropped_attributes_count":    float64(1),
+							"message":                     "This is a log message",
+							"span_id":                     "0102040800000000",
+							"trace_id":                    "08040201000000000000000000000000",
+							"bundle_key":                  "d41d8cd98f00b204e9800998ecf8427e",
+
+							"resource.attributes.resource-attr": "resource-attr-val-1",
+						},
+					},
 					{
 						Thread: testLEventReq.Thread,
 						Log:    testLEventReq.Log,
@@ -482,6 +510,9 @@ func TestConsumeLogsShouldSucceed(t *testing.T) {
 							"span_id":                     "0102040800000000",
 							"trace_id":                    "08040201000000000000000000000000",
 							"bundle_key":                  "d41d8cd98f00b204e9800998ecf8427e",
+
+							"resource.attributes.resource-attr": "resource-attr-val-1",
+							"resource.attributes.serverHost":    "serverHostFromResource",
 						},
 					},
 					{
@@ -498,6 +529,10 @@ func TestConsumeLogsShouldSucceed(t *testing.T) {
 							"span_id":                     "0102040800000000",
 							"trace_id":                    "08040201000000000000000000000000",
 							"bundle_key":                  "d41d8cd98f00b204e9800998ecf8427e",
+
+							"resource.attributes.resource-attr": "resource-attr-val-1",
+							"resource.attributes.host.name":     "serverHostFromResourceHost",
+							"resource.attributes.serverHost":    "serverHostFromResourceServer",
 						},
 					},
 					{
@@ -514,6 +549,54 @@ func TestConsumeLogsShouldSucceed(t *testing.T) {
 							"span_id":                     "0102040800000000",
 							"trace_id":                    "08040201000000000000000000000000",
 							"bundle_key":                  "d41d8cd98f00b204e9800998ecf8427e",
+
+							"resource.attributes.resource-attr": "resource-attr-val-1",
+							"resource.attributes.host.name":     "serverHostFromResourceHost",
+						},
+					},
+					{
+						Thread: testLEventReq.Thread,
+						Log:    testLEventReq.Log,
+						Sev:    testLEventReq.Sev,
+						Ts:     testLEventReq.Ts,
+						Attrs: map[string]interface{}{
+							add_events.AttrOrigServerHost: testServerHost,
+							"app":                         "server",
+							"instance_num":                float64(1),
+							"dropped_attributes_count":    float64(1),
+							"message":                     "This is a log message",
+							"span_id":                     "0102040800000000",
+							"trace_id":                    "08040201000000000000000000000000",
+							"bundle_key":                  "d41d8cd98f00b204e9800998ecf8427e",
+
+							"string":         "string",
+							"double":         2.0,
+							"bool":           true,
+							"empty":          nil,
+							"int":            float64(3),
+							"map_map_empty":  nil,
+							"map_map_string": "map_string",
+							"slice_0":        "slice_string",
+
+							"scope.attributes.string":         "string",
+							"scope.attributes.double":         2.0,
+							"scope.attributes.bool":           true,
+							"scope.attributes.empty":          nil,
+							"scope.attributes.int":            float64(3),
+							"scope.attributes.map.map_empty":  nil,
+							"scope.attributes.map.map_string": "map_string",
+							"scope.attributes.slice.0":        "slice_string",
+
+							"resource.attributes.string":         "string",
+							"resource.attributes.double":         2.0,
+							"resource.attributes.bool":           true,
+							"resource.attributes.empty":          nil,
+							"resource.attributes.int":            float64(3),
+							"resource.attributes.map.map_empty":  nil,
+							"resource.attributes.map.map_string": "map_string",
+							"resource.attributes.slice.0":        "slice_string",
+
+							"resource.attributes.resource-attr": "resource-attr-val-1",
 						},
 					},
 				},

exporter/datasetexporter/traces_exporter_test.go

@@ -210,6 +210,68 @@ func TestBuildEventsFromSpanAttributesCollision(t *testing.T) {
 	assert.Equal(t, expected, was)
 }
 
+func TestBuildEventsFromSpanAttributesDifferentTypes(t *testing.T) {
+	td := ptrace.NewTraces()
+	rs := td.ResourceSpans().AppendEmpty()
+	rss := rs.ScopeSpans().AppendEmpty()
+	span := rss.Spans().AppendEmpty()
+	fillAttributes(span.Attributes())
+	fillAttributes(rss.Scope().Attributes())
+	fillAttributes(rs.Resource().Attributes())
+
+	// sBytes := span.Attributes().PutEmptyBytes("bytes")
+	// sBytes.Append('a')
+	expected := &add_events.EventBundle{
+		Event: &add_events.Event{
+			Thread: "TT",
+			Log:    "LT",
+			Sev:    9,
+			Ts:     "0",
+			Attrs: map[string]interface{}{
+				"sca:schemVer": 1,
+				"sca:schema":   "tracing",
+				"sca:type":     "span",
+
+				"name": "",
+				"kind": "unspecified",
+
+				"start_time_unix_nano": "0",
+				"end_time_unix_nano":   "0",
+				"duration_nano":        "0",
+
+				"span_id":        "",
+				"trace_id":       "",
+				"status_code":    "unset",
+				"status_message": "",
+				"resource_name":  "",
+				"resource_type":  "process",
+				"string":         "string",
+				"double":         2.0,
+				"bool":           true,
+				"empty":          nil,
+				"int":            int64(3),
+
+				"map_map_empty":  nil,
+				"map_map_string": "map_string",
+				"slice_0":        "slice_string",
+			},
+			ServerHost: testServerHost,
+		},
+		Thread: testTThread,
+		Log:    testTLog,
+	}
+	was := buildEventFromSpan(
+		spanBundle{
+			span,
+			rs.Resource(),
+			rss.Scope(),
+		},
+		testServerHost,
+	)
+
+	assert.Equal(t, expected, was)
+}
+
 func TestBuildEventsFromTracesFromTwoSpansSameResourceOneDifferent(t *testing.T) {
 	traces := testdata.GenerateTracesTwoSpansSameResourceOneDifferent()
 	traces.ResourceSpans().At(0).ScopeSpans().At(0).Spans().At(1).Attributes().PutStr("serverHost", "")