pkg/report: get only the thread local reports

After this change we'll see only the reports coming from the same thread.

Author: tarasmadan

pkg/report/linux.go

@@ -165,8 +165,9 @@ func (ctx *linux) Parse(output []byte) *Report {
 	}
 	for questionable := false; ; questionable = true {
 		rep := &Report{
-			Output:   output,
-			StartPos: startPos,
+			Output:    output,
+			StartPos:  startPos,
+			ContextID: context,
 		}
 		endPos, reportEnd, report, prefix := ctx.findReport(output, oops, startPos, context, questionable)
 		rep.EndPos = endPos

pkg/report/report.go

@@ -72,6 +72,10 @@ type Report struct {
 	MachineInfo []byte
 	// If the crash happened in the context of the syz-executor process, Executor will hold more info.
 	Executor *ExecutorInfo
+	// On Linux systems ContextID may be the ThreadID(enabled by CONFIG_PRINTK_CALLER)
+	// or alternatively CpuID.
+	ContextID string
+
 	// reportPrefixLen is length of additional prefix lines that we added before actual crash report.
 	reportPrefixLen int
 	// symbolized is set if the report is symbolized. It prevents double symbolization.
@@ -277,17 +281,41 @@ func IsSuppressed(reporter *Reporter, output []byte) bool {
 		bytes.Contains(output, gceConsoleHangup)
 }
 
-// ParseAll returns all successive reports in output.
-func ParseAll(reporter *Reporter, output []byte) (reports []*Report) {
-	skipPos := 0
+// ParseAll returns all successive reports in output starting from startFrom.
+func ParseAll(reporter *Reporter, output []byte, startFrom int) []*Report {
+	skipPos := startFrom
+	var res []*Report
+	var scanFrom []int
 	for {
 		rep := reporter.ParseFrom(output, skipPos)
 		if rep == nil {
-			return
+			break
 		}
-		reports = append(reports, rep)
+		res = append(res, rep)
+		scanFrom = append(scanFrom, skipPos)
 		skipPos = rep.SkipPos
 	}
+	return fixReports(reporter, res, scanFrom)
+}
+
+func fixReports(reporter *Reporter, reports []*Report, skipPos []int) []*Report {
+	nextContextReportPos := map[string]int{}
+	for i := len(reports) - 1; i >= 0; i-- {
+		rep := reports[i]
+		if rep.Corrupted {
+			continue
+		}
+		nextReportPos := nextContextReportPos[rep.ContextID]
+		nextContextReportPos[rep.ContextID] = rep.StartPos
+		if nextReportPos == 0 {
+			continue
+		}
+		if nextReportPos < rep.EndPos {
+			reports[i] = reporter.ParseFrom(rep.Output[:nextReportPos], skipPos[i])
+			reports[i].Output = rep.Output
+		}
+	}
+	return reports
 }
 
 // GCE console connection sometimes fails with this message.
@@ -958,3 +986,11 @@ func MergeReportBytes(reps []*Report) []byte {
 func SplitReportBytes(data []byte) [][]byte {
 	return bytes.Split(data, []byte(reportSeparator))
 }
+
+func mergeReportContextIDs(reps []*Report) string {
+	var ids []string
+	for _, rep := range reps {
+		ids = append(ids, rep.ContextID)
+	}
+	return strings.Join(ids, ", ")
+}

pkg/report/report_test.go

@@ -44,6 +44,7 @@ type ParseTest struct {
 	HasReport  bool
 	Report     []byte
 	Executor   string
+	ContextIDs string
 	// Only used in report parsing:
 	corruptedReason string
 }
@@ -88,6 +89,9 @@ func (test *ParseTest) Headers(includeFrame bool) []byte {
 	if test.Executor != "" {
 		fmt.Fprintf(buf, "EXECUTOR: %s\n", test.Executor)
 	}
+	if test.ContextIDs != "" {
+		fmt.Fprintf(buf, "CONTEXTS: %s\n", test.ContextIDs)
+	}
 	return buf.Bytes()
 }
 
@@ -96,8 +100,8 @@ func testParseFile(t *testing.T, reporter *Reporter, fn string) {
 	testParseImpl(t, reporter, test)
 }
 
-func parseReport(t *testing.T, reporter *Reporter, fn string) *ParseTest {
-	data, err := os.ReadFile(fn)
+func parseReport(t *testing.T, reporter *Reporter, testFileName string) *ParseTest {
+	data, err := os.ReadFile(testFileName)
 	if err != nil {
 		t.Fatal(err)
 	}
@@ -110,7 +114,7 @@ func parseReport(t *testing.T, reporter *Reporter, fn string) *ParseTest {
 	)
 	phase := phaseHeaders
 	test := &ParseTest{
-		FileName: fn,
+		FileName: testFileName,
 	}
 	prevEmptyLine := false
 	s := bufio.NewScanner(bytes.NewReader(data))
@@ -158,6 +162,7 @@ func parseHeaderLine(t *testing.T, test *ParseTest, ln string) {
 		corruptedPrefix  = "CORRUPTED: "
 		suppressedPrefix = "SUPPRESSED: "
 		executorPrefix   = "EXECUTOR: "
+		contextidPrefix  = "CONTEXTS: "
 	)
 	switch {
 	case strings.HasPrefix(ln, "#"):
@@ -193,6 +198,8 @@ func parseHeaderLine(t *testing.T, test *ParseTest, ln string) {
 		}
 	case strings.HasPrefix(ln, executorPrefix):
 		test.Executor = ln[len(executorPrefix):]
+	case strings.HasPrefix(ln, contextidPrefix):
+		test.ContextIDs = ln[len(contextidPrefix):]
 	default:
 		t.Fatalf("unknown header field %q", ln)
 	}
@@ -545,3 +552,26 @@ func TestSplitReportBytes(t *testing.T) {
 		})
 	}
 }
+
+// TestParseAll's focus points are the ability to:
+// 1. parse multiple reports
+// 2. extract correct ThreadID/CpuID.
+func TestParseAll(t *testing.T) {
+	forEachFile(t, "parse_all", testParseAll)
+}
+
+func testParseAll(t *testing.T, reporter *Reporter, testFileName string) {
+	test := parseReport(t, reporter, testFileName)
+	gotReports := ParseAll(reporter, test.Log, 0)
+	gotIDs := mergeReportContextIDs(gotReports)
+	mergedReport := MergeReportBytes(gotReports)
+	if !bytes.Equal(mergedReport, test.Report) || gotIDs != test.ContextIDs {
+		if *flagUpdate {
+			updateReportTest(t, test, &ParseTest{
+				ContextIDs: gotIDs,
+				Report:     mergedReport})
+		}
+		assert.Equal(t, test.ContextIDs, gotIDs, "extracted wrong Thread or CPU ids")
+		assert.Equal(t, string(test.Report), string(mergedReport), "extracted wrong reports")
+	}
+}