syz-cluster: avoid UUIDs in blob store

syz-cluster/controller/processor.go

@@ -138,7 +138,7 @@ func (sp *SeriesProcessor) handleSession(ctx context.Context, session *db.Sessio
 			app.Errorf("failed to query workflow %q status: %v", session.ID, err)
 			continue
 		}
-		if workflowLog != nil {
+		if len(workflowLog) > 0 {
 			err := sp.updateSessionLog(ctx, session, workflowLog)
 			if err != nil {
 				app.Errorf("failed to update session log: %v", err)
@@ -214,19 +214,12 @@ func (sp *SeriesProcessor) stopRunningTests(ctx context.Context, sessionID strin
 }
 
 func (sp *SeriesProcessor) updateSessionLog(ctx context.Context, session *db.Session, log []byte) error {
+	logURI, err := sp.blobStorage.Write(bytes.NewReader(log), "Session", session.ID, "log")
+	if err != nil {
+		return fmt.Errorf("failed to save the log: %w", err)
+	}
 	return sp.sessionRepo.Update(ctx, session.ID, func(session *db.Session) error {
-		if session.LogURI == "" {
-			path, err := sp.blobStorage.Store(bytes.NewReader(log))
-			if err != nil {
-				return fmt.Errorf("failed to save the log: %w", err)
-			}
-			session.LogURI = path
-		} else {
-			err := sp.blobStorage.Update(session.LogURI, bytes.NewReader(log))
-			if err != nil {
-				return fmt.Errorf("failed to update the log %q: %w", session.LogURI, err)
-			}
-		}
+		session.LogURI = logURI
 		return nil
 	})
 }

syz-cluster/pkg/blob/gcs.go

@@ -7,10 +7,10 @@ import (
 	"context"
 	"fmt"
 	"io"
+	"path"
 	"regexp"
 
 	"github.com/google/syzkaller/pkg/gcs"
-	"github.com/google/uuid"
 )
 
 type gcsDriver struct {
@@ -29,54 +29,39 @@ func NewGCSClient(ctx context.Context, bucket string) (Storage, error) {
 	}, nil
 }
 
-func (gcs *gcsDriver) Store(source io.Reader) (string, error) {
-	object := uuid.NewString()
-	err := gcs.writeObject(object, source)
+func (gcs *gcsDriver) Write(source io.Reader, parts ...string) (string, error) {
+	if len(parts) == 0 {
+		return "", fmt.Errorf("no identifiers for the object were passed to Write")
+	}
+	object := path.Join(gcs.bucket, path.Join(parts...))
+	w, err := gcs.client.FileWriter(object, "", "")
 	if err != nil {
 		return "", err
 	}
-	return gcs.objectURI(object), nil
-}
-
-func (gcs *gcsDriver) Update(uri string, source io.Reader) error {
-	object, err := gcs.objectName(uri)
+	defer w.Close()
+	_, err = io.Copy(w, source)
 	if err != nil {
-		return err
+		return "", err
 	}
-	return gcs.writeObject(object, source)
+	return "gcs://" + object, nil
 }
 
 func (gcs *gcsDriver) Read(uri string) (io.ReadCloser, error) {
-	object, err := gcs.objectName(uri)
+	bucket, object, err := gcs.parseURI(uri)
 	if err != nil {
 		return nil, err
 	}
-	return gcs.client.FileReader(fmt.Sprintf("%s/%s", gcs.bucket, object))
+	return gcs.client.FileReader(path.Join(bucket, object))
 }
 
-var gcsObjectRe = regexp.MustCompile(`^gcs://([\w-]+)/([\w-]+)$`)
+var gcsObjectRe = regexp.MustCompile(`^gcs://([\w-]+)/(.+)$`)
 
-func (gcs *gcsDriver) objectName(uri string) (string, error) {
+func (gcs *gcsDriver) parseURI(uri string) (string, string, error) {
 	match := gcsObjectRe.FindStringSubmatch(uri)
 	if len(match) == 0 {
-		return "", fmt.Errorf("invalid GCS URI")
+		return "", "", fmt.Errorf("invalid GCS URI")
 	} else if match[1] != gcs.bucket {
-		return "", fmt.Errorf("unexpected GCS bucket")
+		return "", "", fmt.Errorf("unexpected GCS bucket")
 	}
-	return match[2], nil
-}
-
-func (gcs *gcsDriver) objectURI(object string) string {
-	return fmt.Sprintf("gcs://%s/%s", gcs.bucket, object)
-}
-
-func (gcs *gcsDriver) writeObject(object string, source io.Reader) error {
-	w, err := gcs.client.FileWriter(fmt.Sprintf("%s/%s", gcs.bucket, object), "", "")
-	if err != nil {
-		return err
-	}
-	defer w.Close()
-
-	_, err = io.Copy(w, source)
-	return err
+	return gcs.bucket, match[2], nil
 }

syz-cluster/pkg/blob/storage.go

@@ -4,21 +4,21 @@
 package blob
 
 import (
+	"encoding/base64"
 	"fmt"
 	"io"
 	"os"
 	"path/filepath"
 	"strings"
-
-	"github.com/google/uuid"
 )
 
 // Storage is not assumed to be used for partciularly large objects (e.g. GB of size),
 // but rather for blobs that risk overwhelming Spanner column size limits.
 type Storage interface {
-	// Store returns a URI to use later.
-	Store(source io.Reader) (string, error)
-	Update(key string, source io.Reader) error
+	// Write stores the object uniquely identified by a set of IDs (parts).
+	// If it already exists, it will be overwritten.
+	// The first argument is the URI which can be used to later retrieve it with Read.
+	Write(source io.Reader, parts ...string) (string, error)
 	Read(uri string) (io.ReadCloser, error)
 }
 
@@ -36,20 +36,19 @@ func NewLocalStorage(baseFolder string) *LocalStorage {
 
 const localStoragePrefix = "local://"
 
-func (ls *LocalStorage) Store(source io.Reader) (string, error) {
-	name := uuid.NewString()
-	err := ls.writeFile(name, source)
+func (ls *LocalStorage) Write(source io.Reader, parts ...string) (string, error) {
+	// A whatever approach that can handle arbitrary inputs.
+	name := base64.StdEncoding.EncodeToString([]byte(filepath.Join(parts...)))
+	file, err := os.Create(filepath.Join(ls.baseFolder, name))
 	if err != nil {
 		return "", err
 	}
-	return localStoragePrefix + name, nil
-}
-
-func (ls *LocalStorage) Update(uri string, source io.Reader) error {
-	if !strings.HasPrefix(uri, localStoragePrefix) {
-		return fmt.Errorf("unsupported URI type")
+	defer file.Close()
+	_, err = io.Copy(file, source)
+	if err != nil {
+		return "", fmt.Errorf("failed to save data: %w", err)
 	}
-	return ls.writeFile(strings.TrimPrefix(uri, localStoragePrefix), source)
+	return localStoragePrefix + name, nil
 }
 
 func (ls *LocalStorage) Read(uri string) (io.ReadCloser, error) {
@@ -61,19 +60,6 @@ func (ls *LocalStorage) Read(uri string) (io.ReadCloser, error) {
 	return os.Open(path)
 }
 
-func (ls *LocalStorage) writeFile(name string, source io.Reader) error {
-	file, err := os.Create(filepath.Join(ls.baseFolder, name))
-	if err != nil {
-		return err
-	}
-	defer file.Close()
-	_, err = io.Copy(file, source)
-	if err != nil {
-		return fmt.Errorf("failed to save data: %w", err)
-	}
-	return nil
-}
-
 func ReadAllBytes(storage Storage, uri string) ([]byte, error) {
 	if uri == "" {
 		return nil, nil

syz-cluster/pkg/blob/storage_test.go

@@ -16,7 +16,7 @@ func TestLocalStorage(t *testing.T) {
 	var uris []string
 	for i := 0; i < 2; i++ {
 		content := fmt.Sprintf("object #%d", i)
-		uri, err := storage.Store(bytes.NewReader([]byte(content)))
+		uri, err := storage.Write(bytes.NewReader([]byte(content)), fmt.Sprint(i))
 		assert.NoError(t, err)
 		uris = append(uris, uri)
 	}

syz-cluster/pkg/service/finding.go

@@ -12,6 +12,7 @@ import (
 	"github.com/google/syzkaller/syz-cluster/pkg/app"
 	"github.com/google/syzkaller/syz-cluster/pkg/blob"
 	"github.com/google/syzkaller/syz-cluster/pkg/db"
+	"github.com/google/uuid"
 )
 
 type FindingService struct {
@@ -27,28 +28,27 @@ func NewFindingService(env *app.AppEnvironment) *FindingService {
 }
 
 func (s *FindingService) Save(ctx context.Context, req *api.NewFinding) error {
-	var reportURI, logURI string
+	finding := &db.Finding{
+		ID:        uuid.NewString(),
+		SessionID: req.SessionID,
+		TestName:  req.TestName,
+		Title:     req.Title,
+	}
 	var err error
 	if len(req.Log) > 0 {
-		logURI, err = s.blobStorage.Store(bytes.NewReader(req.Log))
+		finding.LogURI, err = s.blobStorage.Write(bytes.NewReader(req.Log), "Finding", finding.ID, "log")
 		if err != nil {
 			return fmt.Errorf("failed to save the log: %w", err)
 		}
 	}
 	if len(req.Report) > 0 {
-		reportURI, err = s.blobStorage.Store(bytes.NewReader(req.Report))
+		finding.ReportURI, err = s.blobStorage.Write(bytes.NewReader(req.Report), "Finding", finding.ID, "report")
 		if err != nil {
 			return fmt.Errorf("failed to save the report: %w", err)
 		}
 	}
 	// TODO: if it's not actually addded, the blob records will be orphaned.
-	err = s.findingRepo.Save(ctx, &db.Finding{
-		SessionID: req.SessionID,
-		TestName:  req.TestName,
-		Title:     req.Title,
-		ReportURI: reportURI,
-		LogURI:    logURI,
-	})
+	err = s.findingRepo.Save(ctx, finding)
 	if err == db.ErrFindingExists {
 		// It's ok, just ignore.
 		return nil

syz-cluster/pkg/service/series.go

@@ -13,6 +13,7 @@ import (
 	"github.com/google/syzkaller/syz-cluster/pkg/app"
 	"github.com/google/syzkaller/syz-cluster/pkg/blob"
 	"github.com/google/syzkaller/syz-cluster/pkg/db"
+	"github.com/google/uuid"
 )
 
 // SeriesService is tested in controller/.
@@ -53,6 +54,7 @@ func (s *SeriesService) getSessionSeries(ctx context.Context, sessionID string,
 
 func (s *SeriesService) UploadSeries(ctx context.Context, series *api.Series) (*api.UploadSeriesResp, error) {
 	seriesObj := &db.Series{
+		ID:          uuid.NewString(),
 		ExtID:       series.ExtID,
 		AuthorEmail: series.AuthorEmail,
 		Title:       series.Title,
@@ -66,7 +68,8 @@ func (s *SeriesService) UploadSeries(ctx context.Context, series *api.Series) (*
 		for _, patch := range series.Patches {
 			// In case of errors, we will waste some space, but let's ignore it for simplicity.
 			// Patches are not super big.
-			uri, err := s.blobStorage.Store(bytes.NewReader(patch.Body))
+			uri, err := s.blobStorage.Write(bytes.NewReader(patch.Body),
+				"Series", seriesObj.ID, "Patches", fmt.Sprint(patch.Seq))
 			if err != nil {
 				return nil, fmt.Errorf("failed to upload patch body: %w", err)
 			}

syz-cluster/pkg/service/session.go

@@ -36,9 +36,9 @@ func (s *SessionService) SkipSession(ctx context.Context, sessionID string, skip
 	var triageLogURI string
 	if len(skip.TriageLog) > 0 {
 		var err error
-		triageLogURI, err = s.blobStorage.Store(bytes.NewReader(skip.TriageLog))
+		triageLogURI, err = s.blobStorage.Write(bytes.NewReader(skip.TriageLog), "Session", sessionID, "triage_log")
 		if err != nil {
-			return fmt.Errorf("failed to save the log: %w", err)
+			return fmt.Errorf("failed to save the triage log: %w", err)
 		}
 	}
 	err := s.sessionRepo.Update(ctx, sessionID, func(session *db.Session) error {

syz-cluster/pkg/service/sessiontest.go

@@ -41,7 +41,8 @@ func (s *SessionTestService) Save(ctx context.Context, req *api.TestResult) erro
 	}
 	logURI := entity.LogURI
 	if len(req.Log) > 0 {
-		logURI, err = s.uploadOrUpdate(ctx, logURI, bytes.NewReader(req.Log))
+		logURI, err = s.blobStorage.Write(bytes.NewReader(req.Log),
+			"Session", req.SessionID, "Test", req.TestName, "log")
 		if err != nil {
 			return fmt.Errorf("failed to save the log: %w", err)
 		}
@@ -66,27 +67,11 @@ func (s *SessionTestService) SaveArtifacts(ctx context.Context, sessionID, testN
 	} else if entity == nil {
 		return fmt.Errorf("the test has not been submitted yet")
 	}
-	newArchiveURI, err := s.uploadOrUpdate(ctx, entity.ArtifactsArchiveURI, reader)
+	archiveURI, err := s.blobStorage.Write(reader, "Session", sessionID, "Test", testName, "artifacts")
 	if err != nil {
 		return fmt.Errorf("failed to save the artifacts archive: %w", err)
 	}
 	return s.testRepo.InsertOrUpdate(ctx, entity, func(test *db.SessionTest) {
-		test.ArtifactsArchiveURI = newArchiveURI
+		test.ArtifactsArchiveURI = archiveURI
 	})
 }
-
-func (s *SessionTestService) uploadOrUpdate(ctx context.Context, uri string, reader io.Reader) (string, error) {
-	if uri != "" {
-		err := s.blobStorage.Update(uri, reader)
-		if err != nil {
-			return "", fmt.Errorf("failed to update: %w", err)
-		}
-		return uri, nil
-	}
-	// TODO: it will leak if we fail to save the entity.
-	uri, err := s.blobStorage.Store(reader)
-	if err != nil {
-		return "", fmt.Errorf("failed to save: %w", err)
-	}
-	return uri, nil
-}