syz-ci/manager.go: use context to cancel file transfer instead of pipe CloseWithError

Author: tarasmadan

syz-ci/manager.go

@@ -37,6 +37,7 @@ import (
 	"github.com/google/syzkaller/prog"
 	_ "github.com/google/syzkaller/sys"
 	"github.com/google/syzkaller/sys/targets"
+	"golang.org/x/sync/errgroup"
 )
 
 // This is especially slightly longer than syzkaller rebuild period.
@@ -832,15 +833,19 @@ func (mgr *Manager) uploadBuildAssets(buildInfo *dashapi.Build, assetFolder stri
 	return ret, nil
 }
 
-func (mgr *Manager) httpGET(path string) (resp *http.Response, err error) {
+func (mgr *Manager) httpGET(ctx context.Context, 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))
+	req, err := http.NewRequestWithContext(ctx, "GET", fmt.Sprintf("http://%s%s", addr, path), nil)
+	if err != nil {
+		return nil, err
+	}
+	return client.Do(req)
 }
 
 func (mgr *Manager) uploadCoverReport() error {
@@ -860,7 +865,7 @@ func (mgr *Manager) uploadCoverReport() error {
 	}
 	defer buildSem.Signal()
 
-	resp, err := mgr.httpGET("/cover")
+	resp, err := mgr.httpGET(context.Background(), "/cover")
 	if err != nil {
 		return fmt.Errorf("failed to get report: %w", err)
 	}
@@ -896,7 +901,8 @@ func (mgr *Manager) uploadCoverJSONLToGCS(ctx context.Context, mgrSrc, gcsDest s
 	}
 	defer buildSem.Signal()
 
-	resp, err := mgr.httpGET(mgrSrc)
+	eg, egCtx := errgroup.WithContext(ctx)
+	resp, err := mgr.httpGET(egCtx, mgrSrc)
 	if err != nil {
 		return fmt.Errorf("failed to httpGet %s: %w", mgrSrc, err)
 	}
@@ -909,10 +915,8 @@ func (mgr *Manager) uploadCoverJSONLToGCS(ctx context.Context, mgrSrc, gcsDest s
 	}
 
 	pr, pw := io.Pipe()
-	defer pr.Close()
-	go func() {
-		var closeError error
-		defer func() { pw.CloseWithError(closeError) }()
+	eg.Go(func() error {
+		defer pw.Close()
 		var w io.Writer
 		w = pw
 		if compress {
@@ -923,19 +927,23 @@ func (mgr *Manager) uploadCoverJSONLToGCS(ctx context.Context, mgrSrc, gcsDest s
 		decoder := json.NewDecoder(resp.Body)
 		for decoder.More() {
 			if err := f(w, decoder); err != nil {
-				closeError = fmt.Errorf("callback: %w", err)
-				return
+				return fmt.Errorf("callback: %w", err)
 			}
 		}
-	}()
-	fileName := fmt.Sprintf("%s/%s-%s-%d-%d.jsonl",
-		mgr.mgrcfg.DashboardClient,
-		mgr.name, curTime.Format(time.DateOnly),
-		curTime.Hour(), curTime.Minute())
-	if err := uploadFile(ctx, gcsDest, fileName, pr, publish); err != nil {
-		return fmt.Errorf("failed to uploadFileGCS(): %w", err)
-	}
-	return nil
+		return nil
+	})
+	eg.Go(func() error {
+		defer pr.Close()
+		fileName := fmt.Sprintf("%s/%s-%s-%d-%d.jsonl",
+			mgr.mgrcfg.DashboardClient,
+			mgr.name, curTime.Format(time.DateOnly),
+			curTime.Hour(), curTime.Minute())
+		if err := uploadFile(egCtx, gcsDest, fileName, pr, publish); err != nil {
+			return fmt.Errorf("failed to uploadFileGCS(): %w", err)
+		}
+		return nil
+	})
+	return eg.Wait()
 }
 
 func (mgr *Manager) uploadCoverStat(fuzzingMinutes int) error {
@@ -1043,13 +1051,13 @@ func uploadFile(ctx context.Context, dstPath, name string, file io.Reader, publi
 	log.Logf(0, "uploading %v to %v", name, URLStr)
 	if strings.HasPrefix(URLStr, "http://") ||
 		strings.HasPrefix(URLStr, "https://") {
-		return uploadFileHTTPPut(URLStr, file)
+		return uploadFileHTTPPut(ctx, URLStr, file)
 	}
 	return gcs.UploadFile(ctx, file, URLStr, publish)
 }
 
-func uploadFileHTTPPut(URL string, file io.Reader) error {
-	req, err := http.NewRequest(http.MethodPut, URL, file)
+func uploadFileHTTPPut(ctx context.Context, URL string, file io.Reader) error {
+	req, err := http.NewRequestWithContext(ctx, http.MethodPut, URL, file)
 	if err != nil {
 		return fmt.Errorf("failed to create HTTP PUT request: %w", err)
 	}

syz-ci/manager_test.go

@@ -166,7 +166,7 @@ func TestUploadCoverJSONLToGCS(t *testing.T) {
 			inputJSONL:      "{",
 			inputTime:       time.Time{},
 			wantGCSFileName: "test-bucket/test-namespace/mgr-name-0001-01-01-0-0.jsonl",
-			wantError:       "failed to uploadFileGCS(): io.Copy: callback: cover.ProgramCoverage: unexpected EOF",
+			wantError:       "callback: cover.ProgramCoverage: unexpected EOF",
 		},
 	}