[receiver/otlpjsonfile][bugfix] only add attributes if record/span/metric count is greater than 0 (#38354)

<!--Ex. Fixing a bug - Describe the bug and how this fixes the issue.
Ex. Adding a feature - Explain what this achieves.-->
#### Description

Fix nil pointer dereference due to empty token

<!-- Issue number (e.g. #1234) or full URL to issue, if applicable. -->
#### Link to tracking issue
Fixes
#38289

<!--Describe what testing was performed and which tests were added.-->
#### Testing

Added test which fails without the fix in place.

Author: VihasMakwana

.chloggen/fix-otlpjsonfilereceiver-null-dereference.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: otlpjsonfilereceiver
+
+# A brief description of the change.  Surround your text with quotes ("") if it needs to start with a backtick (`).
+note: Fix nil pointer dereference due to empty token
+
+# Mandatory: One or more tracking issues related to the change. You can use the PR number here if no issue exists.
+issues: [38289]
+
+# (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: []

receiver/otlpjsonfilereceiver/file.go

@@ -88,22 +88,22 @@ func createLogsReceiver(_ context.Context, settings receiver.Settings, configura
 			ctx = obsrecv.StartLogsOp(ctx)
 			var l plog.Logs
 			l, err = logsUnmarshaler.UnmarshalLogs(token)
-			// Appends token.Attributes
-			for i := 0; i < l.ResourceLogs().Len(); i++ {
-				resourceLog := l.ResourceLogs().At(i)
-				for j := 0; j < resourceLog.ScopeLogs().Len(); j++ {
-					scopeLog := resourceLog.ScopeLogs().At(j)
-					for k := 0; k < scopeLog.LogRecords().Len(); k++ {
-						LogRecords := scopeLog.LogRecords().At(k)
-						appendToMap(attributes, LogRecords.Attributes())
-					}
-				}
-			}
 			if err != nil {
 				obsrecv.EndLogsOp(ctx, metadata.Type.String(), 0, err)
 			} else {
 				logRecordCount := l.LogRecordCount()
 				if logRecordCount != 0 {
+					// Appends token.Attributes
+					for i := 0; i < l.ResourceLogs().Len(); i++ {
+						resourceLog := l.ResourceLogs().At(i)
+						for j := 0; j < resourceLog.ScopeLogs().Len(); j++ {
+							scopeLog := resourceLog.ScopeLogs().At(j)
+							for k := 0; k < scopeLog.LogRecords().Len(); k++ {
+								LogRecords := scopeLog.LogRecords().At(k)
+								appendToMap(attributes, LogRecords.Attributes())
+							}
+						}
+					}
 					err = logs.ConsumeLogs(ctx, l)
 				}
 				obsrecv.EndLogsOp(ctx, metadata.Type.String(), logRecordCount, err)
@@ -138,21 +138,21 @@ func createMetricsReceiver(_ context.Context, settings receiver.Settings, config
 			ctx = obsrecv.StartMetricsOp(ctx)
 			var m pmetric.Metrics
 			m, err = metricsUnmarshaler.UnmarshalMetrics(token)
-			// Appends token.Attributes
-			for i := 0; i < m.ResourceMetrics().Len(); i++ {
-				resourceMetric := m.ResourceMetrics().At(i)
-				for j := 0; j < resourceMetric.ScopeMetrics().Len(); j++ {
-					ScopeMetric := resourceMetric.ScopeMetrics().At(j)
-					for k := 0; k < ScopeMetric.Metrics().Len(); k++ {
-						metric := ScopeMetric.Metrics().At(k)
-						appendToMap(attributes, metric.Metadata())
-					}
-				}
-			}
 			if err != nil {
 				obsrecv.EndMetricsOp(ctx, metadata.Type.String(), 0, err)
 			} else {
 				if m.ResourceMetrics().Len() != 0 {
+					// Appends token.Attributes
+					for i := 0; i < m.ResourceMetrics().Len(); i++ {
+						resourceMetric := m.ResourceMetrics().At(i)
+						for j := 0; j < resourceMetric.ScopeMetrics().Len(); j++ {
+							ScopeMetric := resourceMetric.ScopeMetrics().At(j)
+							for k := 0; k < ScopeMetric.Metrics().Len(); k++ {
+								metric := ScopeMetric.Metrics().At(k)
+								appendToMap(attributes, metric.Metadata())
+							}
+						}
+					}
 					err = metrics.ConsumeMetrics(ctx, m)
 				}
 				obsrecv.EndMetricsOp(ctx, metadata.Type.String(), m.MetricCount(), err)
@@ -187,21 +187,21 @@ func createTracesReceiver(_ context.Context, settings receiver.Settings, configu
 			ctx = obsrecv.StartTracesOp(ctx)
 			var t ptrace.Traces
 			t, err = tracesUnmarshaler.UnmarshalTraces(token)
-			// Appends token.Attributes
-			for i := 0; i < t.ResourceSpans().Len(); i++ {
-				resourceSpan := t.ResourceSpans().At(i)
-				for j := 0; j < resourceSpan.ScopeSpans().Len(); j++ {
-					scopeSpan := resourceSpan.ScopeSpans().At(j)
-					for k := 0; k < scopeSpan.Spans().Len(); k++ {
-						spans := scopeSpan.Spans().At(k)
-						appendToMap(attributes, spans.Attributes())
-					}
-				}
-			}
 			if err != nil {
 				obsrecv.EndTracesOp(ctx, metadata.Type.String(), 0, err)
 			} else {
 				if t.ResourceSpans().Len() != 0 {
+					// Appends token.Attributes
+					for i := 0; i < t.ResourceSpans().Len(); i++ {
+						resourceSpan := t.ResourceSpans().At(i)
+						for j := 0; j < resourceSpan.ScopeSpans().Len(); j++ {
+							scopeSpan := resourceSpan.ScopeSpans().At(j)
+							for k := 0; k < scopeSpan.Spans().Len(); k++ {
+								spans := scopeSpan.Spans().At(k)
+								appendToMap(attributes, spans.Attributes())
+							}
+						}
+					}
 					err = traces.ConsumeTraces(ctx, t)
 				}
 				obsrecv.EndTracesOp(ctx, metadata.Type.String(), t.SpanCount(), err)

receiver/otlpjsonfilereceiver/file_test.go

@@ -304,3 +304,56 @@ func TestFileMixedSignals(t *testing.T) {
 	err = pr.Shutdown(context.Background())
 	assert.NoError(t, err)
 }
+
+func TestEmptyLine(t *testing.T) {
+	tempFolder := t.TempDir()
+	factory := NewFactory()
+	cfg := createDefaultConfig().(*Config)
+	cfg.Config.Include = []string{filepath.Join(tempFolder, "*")}
+	cfg.Config.StartAt = "beginning"
+	cs := receivertest.NewNopSettings(metadata.Type)
+	t.Run("metrics receiver", func(t *testing.T) {
+		ms := new(consumertest.MetricsSink)
+		mr, err := factory.CreateMetrics(context.Background(), cs, cfg, ms)
+		assert.NoError(t, err)
+		err = mr.Start(context.Background(), nil)
+		assert.NoError(t, err)
+		defer func() {
+			assert.NoError(t, mr.Shutdown(context.Background()))
+		}()
+		err = os.WriteFile(filepath.Join(tempFolder, "metrics.json"), []byte{'\n', '\n'}, 0o600)
+		assert.NoError(t, err)
+		time.Sleep(1 * time.Second)
+		require.Empty(t, ms.AllMetrics())
+	})
+
+	t.Run("trace receiver", func(t *testing.T) {
+		ts := new(consumertest.TracesSink)
+		tr, err := factory.CreateTraces(context.Background(), cs, cfg, ts)
+		assert.NoError(t, err)
+		err = tr.Start(context.Background(), nil)
+		assert.NoError(t, err)
+		defer func() {
+			assert.NoError(t, tr.Shutdown(context.Background()))
+		}()
+		err = os.WriteFile(filepath.Join(tempFolder, "traces.json"), []byte{'\n', '\n'}, 0o600)
+		assert.NoError(t, err)
+		time.Sleep(1 * time.Second)
+		require.Empty(t, ts.AllTraces())
+	})
+
+	t.Run("log receiver", func(t *testing.T) {
+		ls := new(consumertest.LogsSink)
+		lr, err := factory.CreateLogs(context.Background(), cs, cfg, ls)
+		assert.NoError(t, err)
+		err = lr.Start(context.Background(), nil)
+		assert.NoError(t, err)
+		defer func() {
+			assert.NoError(t, lr.Shutdown(context.Background()))
+		}()
+		err = os.WriteFile(filepath.Join(tempFolder, "logs.json"), []byte{'\n', '\n'}, 0o600)
+		assert.NoError(t, err)
+		time.Sleep(1 * time.Second)
+		require.Empty(t, ls.AllLogs())
+	})
+}