Merge pull request #436 from DataDog/neal.powell/CLOUDS-7101/CLOUDS-7233/CLOUDS-7238/fix.memory.corruption.from.internal.buffer

Fix Data Corruption From Internal Buffer Aliasing

Author: pashneal

forwarder/internal/logs/fixtures/large_logs_buffer_test.json

@@ -89,7 +89,7 @@ func NewLog(logBytes []byte, blob storage.Blob, scrubber Scrubber, originalSize
 	blobNameResourceId := blob.ResourceId()
 	currLog.blobResourceId = blobNameResourceId
 	currLog.byteSize = originalSize
-	currLog.raw = logBytes
+	currLog.raw = slices.Clone(logBytes)
 	currLog.Container = blob.Container.Name
 	currLog.Blob = blob.Name
 

forwarder/internal/logs/models.go

@@ -65,6 +65,9 @@ var (
 
 	//go:embed fixtures/logs_with_level_as_int_or_string.json
 	logsWithLevelAsIntOrStringData []byte
+
+	//go:embed fixtures/large_logs_buffer_test.json
+	largeLogsBufferTestData []byte
 )
 
 func TestParseLogs(t *testing.T) {
@@ -334,3 +337,35 @@ func TestParseActiveDirectoryLogs(t *testing.T) {
 		})
 	}
 }
+
+// Regression [CLOUDS-7233]: Because of a shared reference to the internal scanner buffer,
+// memory corruption would trigger under a particular set of circumstances
+func TestParseLargeLogsBufferReuse(t *testing.T) {
+	t.Parallel()
+
+	// GIVEN: The set of circumstances triggering the bug:
+	// - logs large enough to force buffer overwriting
+	// - pii scrubber rules are not set
+	reader := bytes.NewReader(largeLogsBufferTestData)
+	closer := io.NopCloser(reader)
+	blob := newBlob("/SUBSCRIPTIONS/TEST-SUB/RESOURCEGROUPS/TEST-RG/PROVIDERS/MICROSOFT.TEST/TEST", "insights-logs-test")
+	scrubber := logs.NewPiiScrubber(nil)
+
+	// WHEN: Parsed via logs.Parse()
+	parsedLogsIter, _, err := logs.Parse(closer, blob, scrubber)
+	require.NoError(t, err)
+
+	var collectedLogs []*logs.Log
+	for response := range parsedLogsIter {
+		require.NoError(t, response.Err)
+		collectedLogs = append(collectedLogs, response.ParsedLog)
+	}
+
+	// THEN: Make sure logs aren't corrupted/malformed
+	firstLogContent := string(collectedLogs[0].Content)
+	assert.Contains(t, firstLogContent, "FIRST_LOG_MARKER", "First log Content should contain FIRST_LOG_MARKER")
+	assert.NotContains(t, firstLogContent, "SECOND_LOG_MARKER", "First log Content should NOT contain SECOND_LOG_MARKER (buffer reuse corruption)")
+
+	secondLogContent := string(collectedLogs[1].Content)
+	assert.Contains(t, secondLogContent, "SECOND_LOG_MARKER", "Second log Content should contain SECOND_LOG_MARKER")
+}