syz-ci, pkg/cover: stream coverage from manager to gcs

6 files changed, 196 insertions, 32 deletions

diff --git a/pkg/cover/html.go b/pkg/cover/html.go
index bba3dc07b..6e266ebce 100644
--- a/pkg/cover/html.go
+++ b/pkg/cover/html.go
@@ -211,8 +211,7 @@ func (rg *ReportGenerator) DoRawCoverFiles(w io.Writer, params CoverHandlerParam
 	return nil
 }
 
-type coverageInfo struct {
-	Version   int    `json:"version"`
+type CoverageInfo struct {
 	FilePath  string `json:"file_path"`
 	FuncName  string `json:"func_name"`
 	StartLine int    `json:"sl"`
@@ -245,8 +244,7 @@ func (rg *ReportGenerator) DoCoverJSONL(w io.Writer, params CoverHandlerParams)
 		}
 		pcProgCount := FileByFrame(fm, &frame).lines[frame.StartLine].pcProgCount
 		hitCount := pcProgCount[frame.PC]
-		covInfo := &coverageInfo{
-			Version:   1,
+		covInfo := &CoverageInfo{
 			FilePath:  frame.Name,
 			FuncName:  frame.FuncName,
 			StartLine: frame.Range.StartLine,
diff --git a/pkg/cover/manager_to_ci.go b/pkg/cover/manager_to_ci.go
new file mode 100644
index 000000000..4b6f1d9ef
--- /dev/null
+++ b/pkg/cover/manager_to_ci.go
@@ -0,0 +1,51 @@
+// Copyright 2024 syzkaller project authors. All rights reserved.
+// Use of this source code is governed by Apache 2 LICENSE that can be found in the LICENSE file.
+
+package cover
+
+import (
+	"encoding/json"
+	"fmt"
+	"io"
+)
+
+// CIDetails fields will be added to every CSV line.
+type CIDetails struct {
+	Version        int    `json:"version"`
+	Timestamp      string `json:"timestamp"`
+	FuzzingMinutes int    `json:"fuzzing_minutes"`
+	Arch           string `json:"arch"`
+	BuildID        string `json:"build_id"`
+	Manager        string `json:"manager"`
+	KernelRepo     string `json:"kernel_repo"`
+	KernelBranch   string `json:"kernel_branch"`
+	KernelCommit   string `json:"kernel_commit"`
+}
+
+type dbCoverageRecord struct {
+	CIDetails
+	CoverageInfo
+}
+
+func writeJSLine(w io.Writer, covInfo dbCoverageRecord) error {
+	bs, err := json.Marshal(covInfo)
+	if err != nil {
+		return fmt.Errorf("failed to marshal covInfo: %w", err)
+	}
+	bs = append(bs, '\n')
+	if _, err = w.Write(bs); err != nil {
+		return fmt.Errorf("failed to write js data: %w", err)
+	}
+	return nil
+}
+
+func WriteCIJSONLine(w io.Writer, managerCover CoverageInfo, ciDetails CIDetails) error {
+	dbLine := dbCoverageRecord{
+		CIDetails:    ciDetails,
+		CoverageInfo: managerCover,
+	}
+	if err := writeJSLine(w, dbLine); err != nil {
+		return fmt.Errorf("failed to serialize func line: %w", err)
+	}
+	return nil
+}
diff --git a/pkg/cover/manager_to_ci_test.go b/pkg/cover/manager_to_ci_test.go
new file mode 100644
index 000000000..3bb26beca
--- /dev/null
+++ b/pkg/cover/manager_to_ci_test.go
@@ -0,0 +1,47 @@
+// Copyright 2024 syzkaller project authors. All rights reserved.
+// Use of this source code is governed by Apache 2 LICENSE that can be found in the LICENSE file.
+
+package cover
+
+import (
+	"bytes"
+	"encoding/json"
+	"testing"
+
+	"github.com/stretchr/testify/assert"
+)
+
+func TestWriteCIJSONLine(t *testing.T) {
+	expectedJSON :=
+		`{"version":1,` +
+			`"timestamp":"2014-04-14 00:00:00.000",` +
+			`"fuzzing_minutes":360,` +
+			`"arch":"x86",` +
+			`"build_id":"sample_buildid",` +
+			`"manager":"sample_manager",` +
+			`"kernel_repo":"sample_repo_path",` +
+			`"kernel_branch":"",` +
+			`"kernel_commit":"",` +
+			`"file_path":"main.c",` +
+			`"func_name":"main",` +
+			`"sl":1,"sc":0,"el":1,"ec":-1,` +
+			`"hit_count":1,` +
+			`"inline":false,` +
+			`"pc":12345}
+`
+
+	covInfo := CoverageInfo{}
+	assert.NoError(t, json.Unmarshal(sampleCoverJSON, &covInfo))
+
+	buf := new(bytes.Buffer)
+	assert.NoError(t, WriteCIJSONLine(buf, covInfo, CIDetails{
+		Version:        1,
+		Timestamp:      "2014-04-14 00:00:00.000",
+		FuzzingMinutes: 360,
+		Arch:           "x86",
+		BuildID:        "sample_buildid",
+		Manager:        "sample_manager",
+		KernelRepo:     "sample_repo_path",
+	}))
+	assert.Equal(t, expectedJSON, buf.String())
+}
diff --git a/pkg/cover/report_test.go b/pkg/cover/report_test.go
index e9b224fa5..43769cf46 100644
--- a/pkg/cover/report_test.go
+++ b/pkg/cover/report_test.go
@@ -429,13 +429,13 @@ func checkCSVReport(t *testing.T, CSVReport []byte) {
 	}
 }
 
+var sampleCoverJSON = []byte(`{"file_path":"main.c","func_name":"main",` +
+	`"sl":1,"sc":0,"el":1,"ec":-1,"hit_count":1,"inline":false,"pc":12345}`)
+
 // nolint:lll
 func checkJSONLReport(t *testing.T, r []byte) {
-	expected := []byte(`{"version":1,"file_path":"main.c","func_name":"main",` +
-		`"sl":1,"sc":0,"el":1,"ec":-1,"hit_count":1,"inline":false,"pc":12345}`)
-
 	compacted := new(bytes.Buffer)
-	if err := json.Compact(compacted, expected); err != nil {
+	if err := json.Compact(compacted, sampleCoverJSON); err != nil {
 		t.Errorf("failed to prepare compacted json: %v", err)
 	}
 	compacted.Write([]byte("\n"))
diff --git a/syz-ci/manager.go b/syz-ci/manager.go
index d4021eac5..2f86325e6 100644
--- a/syz-ci/manager.go
+++ b/syz-ci/manager.go
@@ -5,6 +5,7 @@ package main
 
 import (
 	"crypto/sha256"
+	"encoding/json"
 	"errors"
 	"fmt"
 	"io"
@@ -22,6 +23,7 @@ import (
 	"github.com/google/syzkaller/pkg/asset"
 	"github.com/google/syzkaller/pkg/build"
 	"github.com/google/syzkaller/pkg/config"
+	"github.com/google/syzkaller/pkg/cover"
 	"github.com/google/syzkaller/pkg/gcs"
 	"github.com/google/syzkaller/pkg/hash"
 	"github.com/google/syzkaller/pkg/instance"
@@ -171,6 +173,8 @@ var buildSem = instance.NewSemaphore(1)
 // simultaneous env.Test calls.
 var testSem = instance.NewSemaphore(1)
 
+const fuzzingMinutesBeforeCover = 360
+
 func (mgr *Manager) loop() {
 	lastCommit := ""
 	nextBuildTime := time.Now()
@@ -205,10 +209,11 @@ loop:
 			if err := mgr.uploadCoverReport(); err != nil {
 				mgr.Errorf("failed to upload cover report: %v", err)
 			}
-			if mgr.cfg.CorpusUploadPath != "" {
-				if err := mgr.uploadCorpus(); err != nil {
-					mgr.Errorf("failed to upload corpus: %v", err)
-				}
+			if err := mgr.uploadCoverStat(fuzzingMinutesBeforeCover); err != nil {
+				mgr.Errorf("failed to upload coverage stat: %v", err)
+			}
+			if err := mgr.uploadCorpus(); err != nil {
+				mgr.Errorf("failed to upload corpus: %v", err)
 			}
 		}
 
@@ -222,7 +227,7 @@ loop:
 			managerRestartTime = latestInfo.Time
 			mgr.restartManager()
 			if mgr.cmd != nil {
-				artifactUploadTime = time.Now().Add(6 * time.Hour)
+				artifactUploadTime = time.Now().Add(fuzzingMinutesBeforeCover * time.Minute)
 			}
 		}
 
@@ -809,6 +814,17 @@ func (mgr *Manager) uploadBuildAssets(buildInfo *dashapi.Build, assetFolder stri
 	return ret, nil
 }
 
+func (mgr *Manager) httpGET(path string) (resp *http.Response, err error) {
+	addr := mgr.managercfg.HTTP
+	if addr != "" && addr[0] == ':' {
+		addr = "127.0.0.1" + addr // in case addr is ":port"
+	}
+	client := &http.Client{
+		Timeout: time.Hour,
+	}
+	return client.Get(fmt.Sprintf("http://%s%s", addr, path))
+}
+
 func (mgr *Manager) uploadCoverReport() error {
 	directUpload := mgr.managercfg.Cover && mgr.cfg.CoverUploadPath != ""
 	if mgr.storage == nil && !directUpload {
@@ -826,21 +842,13 @@ func (mgr *Manager) uploadCoverReport() error {
 	}
 	defer buildSem.Signal()
 
-	// Get coverage report from manager.
-	addr := mgr.managercfg.HTTP
-	if addr != "" && addr[0] == ':' {
-		addr = "127.0.0.1" + addr // in case addr is ":port"
-	}
-	client := http.Client{
-		Timeout: time.Hour,
-	}
-	resp, err := client.Get(fmt.Sprintf("http://%v/cover", addr))
+	resp, err := mgr.httpGET("/cover")
 	if err != nil {
 		return fmt.Errorf("failed to get report: %w", err)
 	}
 	defer resp.Body.Close()
 	if directUpload {
-		return mgr.uploadFile(mgr.cfg.CoverUploadPath, mgr.name+".html", resp.Body)
+		return mgr.uploadFile(mgr.cfg.CoverUploadPath, mgr.name+".html", resp.Body, true)
 	}
 	// Upload via the asset storage.
 	newAsset, err := mgr.storage.UploadBuildAsset(resp.Body, mgr.name+".html",
@@ -855,16 +863,77 @@ func (mgr *Manager) uploadCoverReport() error {
 	return nil
 }
 
+func (mgr *Manager) uploadCoverStat(fuzzingMinutes int) error {
+	if !mgr.managercfg.Cover || mgr.cfg.CoverPipelinePath == "" {
+		return nil
+	}
+
+	// Report generation consumes 40G RAM. Generate one at a time.
+	// TODO: remove it once #4585 (symbolization tuning) is closed
+	select {
+	case <-buildSem.WaitC():
+	case <-mgr.stop:
+		return nil
+	}
+	defer buildSem.Signal()
+
+	resp, err := mgr.httpGET("/cover?jsonl=1")
+	if err != nil {
+		return fmt.Errorf("failed to get /cover?json=1 report: %w", err)
+	}
+	defer resp.Body.Close()
+
+	curTime := time.Now()
+	pr, pw := io.Pipe()
+	defer pr.Close()
+	go func() {
+		decoder := json.NewDecoder(resp.Body)
+		for decoder.More() {
+			var covInfo cover.CoverageInfo
+			if err := decoder.Decode(&covInfo); err != nil {
+				pw.CloseWithError(fmt.Errorf("failed to decode CoverageInfo: %w", err))
+				return
+			}
+			if err := cover.WriteCIJSONLine(pw, covInfo, cover.CIDetails{
+				Version:        1,
+				Timestamp:      curTime.Format(time.RFC3339Nano),
+				FuzzingMinutes: fuzzingMinutes,
+				Arch:           mgr.lastBuild.Arch,
+				BuildID:        mgr.lastBuild.ID,
+				Manager:        mgr.name,
+				KernelRepo:     mgr.lastBuild.KernelRepo,
+				KernelBranch:   mgr.lastBuild.KernelBranch,
+				KernelCommit:   mgr.lastBuild.KernelCommit,
+			}); err != nil {
+				pw.CloseWithError(fmt.Errorf("failed to write CIJSONLine: %w", err))
+				return
+			}
+		}
+		pw.Close()
+	}()
+	fileName := fmt.Sprintf("%s-%s-%d-%d.jsonl",
+		mgr.name, curTime.Format(time.DateOnly),
+		curTime.Hour(), curTime.Minute())
+	err = mgr.uploadFile(mgr.cfg.CoverPipelinePath, fileName, pr, false)
+	if err != nil {
+		return fmt.Errorf("failed to uploadFileGCS(): %w", err)
+	}
+	return nil
+}
+
 func (mgr *Manager) uploadCorpus() error {
+	if mgr.cfg.CorpusUploadPath == "" {
+		return nil
+	}
 	f, err := os.Open(filepath.Join(mgr.workDir, "corpus.db"))
 	if err != nil {
 		return err
 	}
 	defer f.Close()
-	return mgr.uploadFile(mgr.cfg.CorpusUploadPath, mgr.name+"-corpus.db", f)
+	return mgr.uploadFile(mgr.cfg.CorpusUploadPath, mgr.name+"-corpus.db", f, true)
 }
 
-func (mgr *Manager) uploadFile(dstPath, name string, file io.Reader) error {
+func (mgr *Manager) uploadFile(dstPath, name string, file io.Reader, allowPublishing bool) error {
 	URL, err := url.Parse(dstPath)
 	if err != nil {
 		return fmt.Errorf("failed to parse upload path: %w", err)
@@ -872,18 +941,14 @@ func (mgr *Manager) uploadFile(dstPath, name string, file io.Reader) error {
 	URL.Path = path.Join(URL.Path, name)
 	URLStr := URL.String()
 	log.Logf(0, "uploading %v to %v", name, URLStr)
-	if strings.HasPrefix(URLStr, "gs://") {
-		return uploadFileGCS(strings.TrimPrefix(URLStr, "gs://"), file, mgr.cfg.PublishGCS)
-	}
-	if strings.HasPrefix(URLStr, "http://") ||
-		strings.HasPrefix(URLStr, "https://") {
+	if strings.HasPrefix(URLStr, "http") {
 		return uploadFileHTTPPut(URLStr, file)
 	}
-	// Use GCS as default to maintain backwards compatibility.
-	return uploadFileGCS(URLStr, file, mgr.cfg.PublishGCS)
+	return uploadFileGCS(URLStr, file, allowPublishing && mgr.cfg.PublishGCS)
 }
 
 func uploadFileGCS(URL string, file io.Reader, publish bool) error {
+	URL = strings.TrimPrefix(URL, "gs://")
 	GCS, err := gcs.NewClient()
 	if err != nil {
 		return fmt.Errorf("failed to create GCS client: %w", err)
diff --git a/syz-ci/syz-ci.go b/syz-ci/syz-ci.go
index 7ec88c53e..9c59870ca 100644
--- a/syz-ci/syz-ci.go
+++ b/syz-ci/syz-ci.go
@@ -105,6 +105,9 @@ type Config struct {
 	// Path to upload coverage reports from managers (optional).
 	// Supported protocols: GCS (gs://) and HTTP PUT (http:// or https://).
 	CoverUploadPath string `json:"cover_upload_path"`
+	// Path to upload json coverage reports from managers (optional).
+	// Supported protocol: GCS (gs://)
+	CoverPipelinePath string `json:"cover_pipeline_path"`
 	// Path to upload corpus.db from managers (optional).
 	// Supported protocols: GCS (gs://) and HTTP PUT (http:// or https://).
 	CorpusUploadPath string `json:"corpus_upload_path"`