From 20f157caf33d93b1072a068141c2795c89fe6b38 Mon Sep 17 00:00:00 2001 From: Rex P Date: Thu, 16 Jul 2026 10:56:57 +1000 Subject: [PATCH 1/5] Revert "perf: optimize combine-to-osv bucket listing and avoid partial files.txt upload (#5632)" This reverts commit 038955f85cb51a1763aa40a2cb68a973aae0fccf. --- vulnfeeds/cmd/combine-to-osv/main.go | 76 ++----------------- .../cve/cve5/bulk-converter/main.go | 65 +++------------- .../cmd/converters/cve/nvd-cve-osv/main.go | 38 +--------- vulnfeeds/gcs-tools/gcs.go | 32 +------- vulnfeeds/gcs-tools/gcs_test.go | 38 ---------- 5 files changed, 23 insertions(+), 226 deletions(-) diff --git a/vulnfeeds/cmd/combine-to-osv/main.go b/vulnfeeds/cmd/combine-to-osv/main.go index 60f3ab72ae1..1427f35a559 100644 --- a/vulnfeeds/cmd/combine-to-osv/main.go +++ b/vulnfeeds/cmd/combine-to-osv/main.go @@ -4,7 +4,6 @@ package main import ( "cmp" "context" - "errors" "flag" "fmt" "io" @@ -17,7 +16,6 @@ import ( "strings" "sync" "sync/atomic" - "time" "cloud.google.com/go/storage" "github.com/google/osv/vulnfeeds/conversion" @@ -32,10 +30,9 @@ import ( ) const ( - defaultOSVOutputPath = "osv-output" - defaultCVE5Path = "cve5" - defaultNVDOSVPath = "nvd" - defaultListFileMaxAge = 2 * 24 * time.Hour + defaultOSVOutputPath = "osv-output" + defaultCVE5Path = "cve5" + defaultNVDOSVPath = "nvd" ) // CVEWorkItem represents a unit of work for a single CVE. @@ -60,71 +57,11 @@ func cveIDFromPath(p string) models.CVEID { return "" } -// readListFile reads the list file (usually files.txt) from GCS. -// The list file is expected to contain newline-separated GCS object names -// relative to the bucket root (e.g. "prefix/CVE-2023-0001.json"). -// It returns a slice of full GCS paths (e.g. "gs://bucket/prefix/CVE-2023-0001.json"). -func readListFile(ctx context.Context, listObj *storage.ObjectHandle, bucketName string) ([]string, error) { - rc, err := listObj.NewReader(ctx) - if err != nil { - return nil, fmt.Errorf("failed to open reader: %w", err) - } - defer rc.Close() - - content, err := io.ReadAll(rc) - if err != nil { - return nil, fmt.Errorf("failed to read content: %w", err) - } - - lines := strings.Split(string(content), "\n") - fullPaths := make([]string, 0, len(lines)) - for _, line := range lines { - line = strings.TrimSpace(line) - if line == "" { - continue - } - fullPaths = append(fullPaths, fmt.Sprintf("gs://%s/%s", bucketName, line)) - } - - return fullPaths, nil -} - -func listObjects(ctx context.Context, client *storage.Client, pathStr string, maxAge time.Duration) ([]string, error) { +func listObjects(ctx context.Context, client *storage.Client, pathStr string) ([]string, error) { if strings.HasPrefix(pathStr, "gs://") { trimmed := strings.TrimPrefix(pathStr, "gs://") bucketName, prefix, _ := strings.Cut(trimmed, "/") bucket := client.Bucket(bucketName) - - listObjectName := gcs.FileListFilename - if prefix != "" { - listObjectName = strings.TrimSuffix(prefix, "/") + "/" + gcs.FileListFilename - } - - listObj := bucket.Object(listObjectName) - attrs, err := listObj.Attrs(ctx) - useListFile := false - if err == nil { - // Check age - age := time.Since(attrs.Updated) - if age <= maxAge { - useListFile = true - } else { - logger.Error("List file is too old (generators might be failing)", slog.String("bucket", bucketName), slog.String("prefix", prefix), slog.Duration("age", age)) - } - } else if !errors.Is(err, storage.ErrObjectNotExist) { - logger.Warn("Failed to get list file attributes", slog.String("bucket", bucketName), slog.String("prefix", prefix), slog.Any("err", err)) - } - - if useListFile { - logger.Info("Using list file", slog.String("bucket", bucketName), slog.String("prefix", prefix)) - fullPaths, err := readListFile(ctx, listObj, bucketName) - if err == nil { - return fullPaths, nil - } - logger.Error("Failed to read list file, falling back to bucket listing", slog.String("bucket", bucketName), slog.String("prefix", prefix), slog.Any("err", err)) - } - - logger.Info("Falling back to bucket listing", slog.String("bucket", bucketName), slog.String("prefix", prefix)) objs, err := gcs.ListBucketObjects(ctx, bucket, prefix) if err != nil { return nil, err @@ -256,7 +193,6 @@ func main() { uploadToGCS := flag.Bool("upload-to-gcs", false, "If true, upload to GCS bucket instead of writing to local disk.") numWorkers := flag.Int("workers", 64, "Number of workers to process records") syncDeletions := flag.Bool("sync-deletions", false, "If false, do not delete files in bucket that are not local") - listFileMaxAge := flag.Duration("list-file-max-age", defaultListFileMaxAge, "The maximum age of the list file before falling back to listing the bucket.") flag.Parse() err := os.MkdirAll(*osvOutputPath, 0755) @@ -281,7 +217,7 @@ func main() { go func() { defer listObjWg.Done() logger.Info("Listing CVE5 objects", slog.String("path", *cve5Path)) - cve5Files, cve5ListErr = listObjects(ctx, client, *cve5Path, *listFileMaxAge) + cve5Files, cve5ListErr = listObjects(ctx, client, *cve5Path) if cve5ListErr != nil { logger.Fatal("Failed to list CVE5 objects", slog.Any("err", cve5ListErr)) } @@ -290,7 +226,7 @@ func main() { go func() { defer listObjWg.Done() logger.Info("Listing NVD objects", slog.String("path", *nvdPath)) - nvdFiles, nvdListErr = listObjects(ctx, client, *nvdPath, *listFileMaxAge) + nvdFiles, nvdListErr = listObjects(ctx, client, *nvdPath) if nvdListErr != nil { logger.Fatal("Failed to list NVD objects", slog.Any("err", nvdListErr)) } diff --git a/vulnfeeds/cmd/converters/cve/cve5/bulk-converter/main.go b/vulnfeeds/cmd/converters/cve/cve5/bulk-converter/main.go index ff0f950b2ca..365470e71ce 100644 --- a/vulnfeeds/cmd/converters/cve/cve5/bulk-converter/main.go +++ b/vulnfeeds/cmd/converters/cve/cve5/bulk-converter/main.go @@ -8,7 +8,6 @@ import ( "flag" "log/slog" "os" - "path" "path/filepath" "slices" "strconv" @@ -23,12 +22,10 @@ import ( "github.com/google/osv/vulnfeeds/utility/logger" ) -const defaultStartYear = "2022" - var ( repoDir = flag.String("cve5-repo", "cvelistV5", "CVEListV5 directory path") localOutputDir = flag.String("out-dir", "cve5", "Path to output results.") - startYear = flag.String("start-year", defaultStartYear, "The first in scope year to process.") + startYear = flag.String("start-year", "2022", "The first in scope year to process.") workers = flag.Int("workers", 10, "The number of concurrent workers to use for processing CVEs.") gcsWorkers = flag.Int("gcs-workers", 30, "The number of concurrent workers to use for GCS uploads.") cnaDenyList = flag.String("cna-denylist", "", "A comma-separated list of CNAs to skip. If not provided, defaults to cna_denylist.txt.") @@ -46,20 +43,6 @@ func main() { logger.InitGlobalLogger() defer logger.Close() - startYearInt, err := strconv.Atoi(*startYear) - defaultStartYearInt, _ := strconv.Atoi(defaultStartYear) - isPartial := false - if err != nil { - logger.Error("Invalid start-year, assuming partial", slog.String("start-year", *startYear)) - isPartial = true - } else if startYearInt > defaultStartYearInt { - isPartial = true - } - - if isPartial { - logger.Info("Partial run detected (start-year > " + defaultStartYear + "), will skip files.txt upload") - } - logger.Info("Commencing CVE to OSV conversion run") if err := os.MkdirAll(*localOutputDir, 0755); err != nil { logger.Fatal("Failed to create local output directory", slog.Any("err", err)) @@ -87,29 +70,20 @@ func main() { if err != nil { logger.Fatal("Failed to initialize GCS upload pool", slog.Any("err", err)) } + defer gcsHelper.CloseAndWait() logger.Info("GCS Upload Pool initialized", slog.String("bucket", *outputBucket)) } - outputFilesChan := make(chan string) - var collectorWg sync.WaitGroup - var outputFiles []string - collectorWg.Add(1) - go func() { - defer collectorWg.Done() - for f := range outputFilesChan { - outputFiles = append(outputFiles, f) - } - }() - // Start the worker pool. for range *workers { wg.Add(1) - go worker(&wg, jobs, gcsHelper, *localOutputDir, cnaList, *rejectFailed, outputFilesChan) + go worker(&wg, jobs, gcsHelper, *localOutputDir, cnaList, *rejectFailed) } // Discover files and send them to the workers. logger.Info("Starting conversion of CVEs...") currentYear := time.Now().Year() + startYearInt, _ := strconv.Atoi(*startYear) for year := startYearInt; year <= currentYear; year++ { year := strconv.Itoa(year) @@ -138,37 +112,22 @@ func main() { close(jobs) wg.Wait() - close(outputFilesChan) - collectorWg.Wait() - - if *uploadToGCS && gcsHelper != nil { - gcsHelper.CloseAndWait() - if !isPartial { - logger.Info("Uploading output file list to GCS...") - if err := gcs.UploadFileList(ctx, *outputBucket, *gcsPrefix, outputFiles); err != nil { - logger.Error("Failed to upload output file list", slog.Any("err", err)) - } - } else { - logger.Info("Skipping files.txt upload due to partial run") - } - } - logger.Info("CVE5 Conversion run complete") } // worker is a function that processes CVE files from the jobs channel. -func worker(wg *sync.WaitGroup, jobs <-chan string, gcsHelper *gcs.Helper, outDir string, cnas []string, rejectFailed bool, outputFilesChan chan<- string) { +func worker(wg *sync.WaitGroup, jobs <-chan string, gcsHelper *gcs.Helper, outDir string, cnas []string, rejectFailed bool) { defer wg.Done() - for cvePath := range jobs { - data, err := os.ReadFile(cvePath) + for path := range jobs { + data, err := os.ReadFile(path) if err != nil { - logger.Info("Failed to read file", slog.String("path", cvePath), slog.Any("err", err)) + logger.Info("Failed to read file", slog.String("path", path), slog.Any("err", err)) continue } var cve models.CVE5 if err := json.Unmarshal(data, &cve); err != nil { - logger.Info("Failed to unmarshal JSON", slog.String("path", cvePath), slog.Any("err", err)) + logger.Info("Failed to unmarshal JSON", slog.String("path", path), slog.Any("err", err)) continue } @@ -180,9 +139,9 @@ func worker(wg *sync.WaitGroup, jobs <-chan string, gcsHelper *gcs.Helper, outDi sourceLink := "" baseDirCVEList := "cves/" // The base folder for the CVEListV5 repository. - idx := strings.Index(cvePath, baseDirCVEList) + idx := strings.Index(path, baseDirCVEList) if idx != -1 { - relPath := cvePath[idx:] + relPath := path[idx:] sourceLink = "https://github.com/CVEProject/cvelistV5/tree/main/" + relPath } @@ -194,8 +153,6 @@ func worker(wg *sync.WaitGroup, jobs <-chan string, gcsHelper *gcs.Helper, outDi logger.Info("Queueing OSV record for "+string(cveID), slog.String("cve", string(cveID))) if err := writer.UploadVulnIfChangedAsync(gcsHelper, *gcsPrefix, vuln.Vulnerability); err != nil { logger.Error("Failed to queue vulnerability upload", slog.String("cve", string(cveID)), slog.Any("err", err)) - } else { - outputFilesChan <- path.Join(*gcsPrefix, string(cveID)+".json") } if err := writer.UploadMetricsToGCSAsync(gcsHelper, *gcsPrefix, cveID, metrics); err != nil { diff --git a/vulnfeeds/cmd/converters/cve/nvd-cve-osv/main.go b/vulnfeeds/cmd/converters/cve/nvd-cve-osv/main.go index 56f1172fcab..6301252985e 100644 --- a/vulnfeeds/cmd/converters/cve/nvd-cve-osv/main.go +++ b/vulnfeeds/cmd/converters/cve/nvd-cve-osv/main.go @@ -8,7 +8,6 @@ import ( "log/slog" "net/http" "os" - "path" "path/filepath" "regexp" "runtime/pprof" @@ -27,8 +26,6 @@ import ( "github.com/google/osv/vulnfeeds/vulns" ) -const defaultStartYear = 2016 - var ( jsonPath = flag.String("nvd-json", "", "Path to NVD CVE JSON to examine.") jsonDir = flag.String("nvd-json-dir", "", "Path to directory containing NVD CVE JSON files to examine.") @@ -42,7 +39,7 @@ var ( uploadToGCS = flag.Bool("upload-to-gcs", false, "If true, upload to GCS bucket instead of writing to local disk.") outputBucket = flag.String("output-bucket", "osv-test-cve-osv-conversion", "The GCS bucket to write to.") gcsPrefix = flag.String("gcs-prefix", "nvd-osv", "The prefix within the GCS bucket.") - startYear = flag.Int("start-year", defaultStartYear, "The first in scope year to process. If 0, process all years.") + startYear = flag.Int("start-year", 2016, "The first in scope year to process. If 0, process all years.") ) var ( @@ -74,11 +71,6 @@ func main() { logger.InitGlobalLogger() defer logger.Close() - isPartial := *startYear > defaultStartYear - if isPartial { - logger.Info("Partial run detected (start-year > " + strconv.Itoa(defaultStartYear) + "), will skip files.txt upload") - } - var files []string if *jsonDir != "" { matches, err := filepath.Glob(filepath.Join(*jsonDir, "nvdcve-2.0-*.json")) @@ -138,26 +130,16 @@ func main() { if err != nil { logger.Fatal("Failed to initialize GCS upload pool", slog.Any("err", err)) } + defer gcsHelper.CloseAndWait() logger.Info("GCS Upload Pool initialized", slog.String("bucket", *outputBucket)) } - outputFilesChan := make(chan string) - var collectorWg sync.WaitGroup - var outputFiles []string - collectorWg.Add(1) - go func() { - defer collectorWg.Done() - for f := range outputFilesChan { - outputFiles = append(outputFiles, f) - } - }() - jobs := make(chan models.NVDCVE, *workers) var wg sync.WaitGroup for range *workers { wg.Add(1) - go worker(&wg, jobs, gcsHelper, *outDir, vpRepoCache, repoTagsCache, outputFilesChan) + go worker(&wg, jobs, gcsHelper, *outDir, vpRepoCache, repoTagsCache) } for _, file := range files { @@ -182,21 +164,11 @@ func main() { close(jobs) wg.Wait() - close(outputFilesChan) - collectorWg.Wait() timesBlocked := int64(0) if gcsHelper != nil { timesBlocked = gcsHelper.GetTimesBlocked() gcsHelper.CloseAndWait() - if !isPartial { - logger.Info("Uploading output file list to GCS...") - if err := gcs.UploadFileList(ctx, *outputBucket, *gcsPrefix, outputFiles); err != nil { - logger.Error("Failed to upload output file list", slog.Any("err", err)) - } - } else { - logger.Info("Skipping files.txt upload due to partial run") - } } logger.Info("Conversion Stats", slog.Uint64("total_processed", totalConversionsCount.Load()), @@ -227,7 +199,7 @@ func processCVE(cve models.NVDCVE, vpRepoCache *c.VPRepoCache, repoTagsCache git return nvd.CVEToOSV(cve, repos, vpRepoCache, repoTagsCache, metrics) } -func worker(wg *sync.WaitGroup, jobs <-chan models.NVDCVE, gcsHelper *gcs.Helper, outDir string, vpRepoCache *c.VPRepoCache, repoTagsCache git.RepoTagsCache, outputFilesChan chan<- string) { +func worker(wg *sync.WaitGroup, jobs <-chan models.NVDCVE, gcsHelper *gcs.Helper, outDir string, vpRepoCache *c.VPRepoCache, repoTagsCache git.RepoTagsCache) { defer wg.Done() for cve := range jobs { vuln, metrics, outcome := processCVE(cve, vpRepoCache, repoTagsCache) @@ -273,8 +245,6 @@ func worker(wg *sync.WaitGroup, jobs <-chan models.NVDCVE, gcsHelper *gcs.Helper if vuln != nil { if err := writer.UploadVulnIfChangedAsync(gcsHelper, *gcsPrefix, vuln.Vulnerability); err != nil { logger.Error("Failed to queue vulnerability upload", slog.String("cve", vuln.Id), slog.Any("err", err)) - } else { - outputFilesChan <- path.Join(*gcsPrefix, vuln.Id+".json") } } if *outputMetrics && metrics != nil { diff --git a/vulnfeeds/gcs-tools/gcs.go b/vulnfeeds/gcs-tools/gcs.go index d54791c6628..ef34e6017d0 100644 --- a/vulnfeeds/gcs-tools/gcs.go +++ b/vulnfeeds/gcs-tools/gcs.go @@ -17,12 +17,10 @@ import ( "github.com/google/osv/vulnfeeds/utility/logger" "golang.org/x/sync/errgroup" "google.golang.org/api/iterator" - "google.golang.org/api/option" ) const ( - hashMetadataKey = "sha256-hash" // hashMetadataKey is the key for the sha256 hash in the GCS object metadata. - FileListFilename = "files.txt" + hashMetadataKey = "sha256-hash" // hashMetadataKey is the key for the sha256 hash in the GCS object metadata. ) type Helper struct { @@ -223,7 +221,7 @@ func DownloadBucket(ctx context.Context, bkt *storage.BucketHandle, prefix strin return nil } -// ListBucketObjects lists the names of all objects in a Google Cloud Storage bucket. +// listBucketObjects lists the names of all objects in a Google Cloud Storage bucket. // It does not download the file contents. func ListBucketObjects(ctx context.Context, bucket *storage.BucketHandle, prefix string) ([]string, error) { it := bucket.Objects(ctx, &storage.Query{Prefix: prefix}) @@ -241,29 +239,3 @@ func ListBucketObjects(ctx context.Context, bucket *storage.BucketHandle, prefix return filenames, nil } - -// UploadFileList uploads a list of files as a single text file to GCS. -func UploadFileList(ctx context.Context, bucketName string, prefix string, files []string, opts ...option.ClientOption) error { - client, err := storage.NewClient(ctx, opts...) - if err != nil { - return fmt.Errorf("storage.NewClient: %w", err) - } - defer client.Close() - - bkt := client.Bucket(bucketName) - - var objectName string - if prefix == "" { - objectName = FileListFilename - } else { - objectName = strings.TrimSuffix(prefix, "/") + "/" + FileListFilename - } - - content := strings.Join(files, "\n") - if len(files) > 0 { - content += "\n" // Add trailing newline - } - reader := strings.NewReader(content) - - return UploadToGCS(ctx, bkt, objectName, reader, "text/plain", nil) -} diff --git a/vulnfeeds/gcs-tools/gcs_test.go b/vulnfeeds/gcs-tools/gcs_test.go index f4e888f9514..af8a2c7f94f 100644 --- a/vulnfeeds/gcs-tools/gcs_test.go +++ b/vulnfeeds/gcs-tools/gcs_test.go @@ -8,7 +8,6 @@ import ( "testing" "github.com/fsouza/fake-gcs-server/fakestorage" - "google.golang.org/api/option" ) func TestUploadToGCS(t *testing.T) { @@ -209,40 +208,3 @@ func TestDownloadBucket(t *testing.T) { } }) } - -func TestUploadFileList(t *testing.T) { - server := fakestorage.NewServer([]fakestorage.Object{}) - t.Cleanup(server.Stop) - - t.Log("Server URL:", server.URL()) - - bucketName := "test-bucket" - prefix := "test-prefix" - files := []string{ - "test-prefix/file1.json", - "test-prefix/file2.json", - } - - ctx := context.Background() - // Create bucket first (fake server needs it) - client := server.Client() - bkt := client.Bucket(bucketName) - if err := bkt.Create(ctx, "project", nil); err != nil { - t.Fatalf("failed to create bucket: %v", err) - } - - err := UploadFileList(ctx, bucketName, prefix, files, option.WithHTTPClient(server.HTTPClient())) - if err != nil { - t.Fatalf("UploadFileList failed: %v", err) - } - - obj, err := server.GetObject(bucketName, "test-prefix/files.txt") - if err != nil { - t.Fatalf("failed to get object: %v", err) - } - - expectedContent := "test-prefix/file1.json\ntest-prefix/file2.json\n" - if string(obj.Content) != expectedContent { - t.Errorf("expected content %q, got %q", expectedContent, string(obj.Content)) - } -} From 054230ddecfa094167e8e6b95fad8fdea5717104 Mon Sep 17 00:00:00 2001 From: Rex P Date: Thu, 16 Jul 2026 13:23:05 +1000 Subject: [PATCH 2/5] Speedup --- osv.dev.code-workspace | 17 +++++++++ vulnfeeds/cmd/combine-to-osv/main.go | 57 ++++++++++++++++++++++++++-- vulnfeeds/gcs-tools/gcs.go | 16 +++++++- 3 files changed, 85 insertions(+), 5 deletions(-) create mode 100644 osv.dev.code-workspace diff --git a/osv.dev.code-workspace b/osv.dev.code-workspace new file mode 100644 index 00000000000..2231d56d6c1 --- /dev/null +++ b/osv.dev.code-workspace @@ -0,0 +1,17 @@ +{ + "folders": [ + { + "path": "." + }, + { + "path": "go" + }, + { + "path": "vulnfeeds" + }, + { + "path": "go/cmd/tools/reimport-tui" + } + ], + "settings": {} +} \ No newline at end of file diff --git a/vulnfeeds/cmd/combine-to-osv/main.go b/vulnfeeds/cmd/combine-to-osv/main.go index 1427f35a559..3091b559b97 100644 --- a/vulnfeeds/cmd/combine-to-osv/main.go +++ b/vulnfeeds/cmd/combine-to-osv/main.go @@ -16,6 +16,7 @@ import ( "strings" "sync" "sync/atomic" + "time" "cloud.google.com/go/storage" "github.com/google/osv/vulnfeeds/conversion" @@ -25,6 +26,7 @@ import ( "github.com/google/osv/vulnfeeds/utility" "github.com/google/osv/vulnfeeds/utility/logger" "github.com/ossf/osv-schema/bindings/go/osvschema" + "golang.org/x/sync/errgroup" "google.golang.org/protobuf/encoding/protojson" "google.golang.org/protobuf/types/known/structpb" ) @@ -33,6 +35,7 @@ const ( defaultOSVOutputPath = "osv-output" defaultCVE5Path = "cve5" defaultNVDOSVPath = "nvd" + parallelStartYear = 2018 ) // CVEWorkItem represents a unit of work for a single CVE. @@ -61,13 +64,61 @@ func listObjects(ctx context.Context, client *storage.Client, pathStr string) ([ if strings.HasPrefix(pathStr, "gs://") { trimmed := strings.TrimPrefix(pathStr, "gs://") bucketName, prefix, _ := strings.Cut(trimmed, "/") + prefix = strings.TrimSuffix(prefix, "/") bucket := client.Bucket(bucketName) - objs, err := gcs.ListBucketObjects(ctx, bucket, prefix) - if err != nil { + + currentYear := time.Now().Year() + var queries []*storage.Query + + // Before 2018 query + before2018Query := &storage.Query{} + startSeg := "CVE-" + endSeg := fmt.Sprintf("CVE-%d-", parallelStartYear) + if prefix != "" { + before2018Query.Prefix = prefix + "/" + before2018Query.StartOffset = prefix + "/" + startSeg + before2018Query.EndOffset = prefix + "/" + endSeg + } else { + before2018Query.StartOffset = startSeg + before2018Query.EndOffset = endSeg + } + queries = append(queries, before2018Query) + + // Parallel queries for each year from 2018 to currentYear + for year := parallelStartYear; year <= currentYear; year++ { + yearPrefix := fmt.Sprintf("CVE-%d-", year) + if prefix != "" { + yearPrefix = prefix + "/" + yearPrefix + } + queries = append(queries, &storage.Query{Prefix: yearPrefix}) + } + + objsChan := make(chan []string, len(queries)) + g, ctx := errgroup.WithContext(ctx) + + for _, q := range queries { + g.Go(func() error { + objs, err := gcs.ListBucketObjectsQuery(ctx, bucket, q) + if err != nil { + return err + } + objsChan <- objs + return nil + }) + } + + if err := g.Wait(); err != nil { return nil, err } + close(objsChan) + + var allObjs []string + for objs := range objsChan { + allObjs = append(allObjs, objs...) + } + var fullPaths []string - for _, obj := range objs { + for _, obj := range allObjs { if strings.HasSuffix(obj, "/") || !strings.HasSuffix(obj, ".json") || strings.HasSuffix(obj, ".metrics.json") { continue } diff --git a/vulnfeeds/gcs-tools/gcs.go b/vulnfeeds/gcs-tools/gcs.go index ef34e6017d0..2b32b241f5f 100644 --- a/vulnfeeds/gcs-tools/gcs.go +++ b/vulnfeeds/gcs-tools/gcs.go @@ -221,10 +221,22 @@ func DownloadBucket(ctx context.Context, bkt *storage.BucketHandle, prefix strin return nil } -// listBucketObjects lists the names of all objects in a Google Cloud Storage bucket. +// ListBucketObjects lists the names of all objects in a Google Cloud Storage bucket. // It does not download the file contents. func ListBucketObjects(ctx context.Context, bucket *storage.BucketHandle, prefix string) ([]string, error) { - it := bucket.Objects(ctx, &storage.Query{Prefix: prefix}) + return ListBucketObjectsQuery(ctx, bucket, &storage.Query{Prefix: prefix}) +} + +// ListBucketObjectsQuery lists the names of all objects in a Google Cloud Storage bucket matching the query. +// It optimizes the request by only retrieving the Name attribute. +func ListBucketObjectsQuery(ctx context.Context, bucket *storage.BucketHandle, query *storage.Query) ([]string, error) { + if query == nil { + query = &storage.Query{} + } + if err := query.SetAttrSelection([]string{"Name"}); err != nil { + return nil, fmt.Errorf("failed to set attribute selection: %w", err) + } + it := bucket.Objects(ctx, query) var filenames []string for { attrs, err := it.Next() From 179c20af255b33614d0bcd1cc7c913497e3eb29a Mon Sep 17 00:00:00 2001 From: Rex P Date: Thu, 16 Jul 2026 14:16:21 +1000 Subject: [PATCH 3/5] Fancy abstractions and tests --- vulnfeeds/cmd/combine-to-osv/main.go | 53 ++------------- vulnfeeds/conversion/writer/writer.go | 8 ++- vulnfeeds/gcs-tools/gcs.go | 85 +++++++++++++++++++++++-- vulnfeeds/gcs-tools/gcs_test.go | 92 +++++++++++++++++++++++++++ 4 files changed, 184 insertions(+), 54 deletions(-) diff --git a/vulnfeeds/cmd/combine-to-osv/main.go b/vulnfeeds/cmd/combine-to-osv/main.go index 3091b559b97..8992181c5bd 100644 --- a/vulnfeeds/cmd/combine-to-osv/main.go +++ b/vulnfeeds/cmd/combine-to-osv/main.go @@ -26,7 +26,6 @@ import ( "github.com/google/osv/vulnfeeds/utility" "github.com/google/osv/vulnfeeds/utility/logger" "github.com/ossf/osv-schema/bindings/go/osvschema" - "golang.org/x/sync/errgroup" "google.golang.org/protobuf/encoding/protojson" "google.golang.org/protobuf/types/known/structpb" ) @@ -64,62 +63,22 @@ func listObjects(ctx context.Context, client *storage.Client, pathStr string) ([ if strings.HasPrefix(pathStr, "gs://") { trimmed := strings.TrimPrefix(pathStr, "gs://") bucketName, prefix, _ := strings.Cut(trimmed, "/") - prefix = strings.TrimSuffix(prefix, "/") bucket := client.Bucket(bucketName) currentYear := time.Now().Year() - var queries []*storage.Query - - // Before 2018 query - before2018Query := &storage.Query{} - startSeg := "CVE-" - endSeg := fmt.Sprintf("CVE-%d-", parallelStartYear) - if prefix != "" { - before2018Query.Prefix = prefix + "/" - before2018Query.StartOffset = prefix + "/" + startSeg - before2018Query.EndOffset = prefix + "/" + endSeg - } else { - before2018Query.StartOffset = startSeg - before2018Query.EndOffset = endSeg - } - queries = append(queries, before2018Query) - - // Parallel queries for each year from 2018 to currentYear + var breakdowns []string for year := parallelStartYear; year <= currentYear; year++ { - yearPrefix := fmt.Sprintf("CVE-%d-", year) - if prefix != "" { - yearPrefix = prefix + "/" + yearPrefix - } - queries = append(queries, &storage.Query{Prefix: yearPrefix}) + breakdowns = append(breakdowns, fmt.Sprintf("CVE-%d-", year)) } - objsChan := make(chan []string, len(queries)) - g, ctx := errgroup.WithContext(ctx) - - for _, q := range queries { - g.Go(func() error { - objs, err := gcs.ListBucketObjectsQuery(ctx, bucket, q) - if err != nil { - return err - } - objsChan <- objs - return nil - }) - } - - if err := g.Wait(); err != nil { + objs, err := gcs.ListObjectsFast(ctx, bucket, prefix, breakdowns) + if err != nil { return nil, err } - close(objsChan) - - var allObjs []string - for objs := range objsChan { - allObjs = append(allObjs, objs...) - } var fullPaths []string - for _, obj := range allObjs { - if strings.HasSuffix(obj, "/") || !strings.HasSuffix(obj, ".json") || strings.HasSuffix(obj, ".metrics.json") { + for _, obj := range objs { + if !strings.HasSuffix(obj, ".json") || strings.HasSuffix(obj, ".metrics.json") { continue } fullPaths = append(fullPaths, fmt.Sprintf("gs://%s/%s", bucketName, obj)) diff --git a/vulnfeeds/conversion/writer/writer.go b/vulnfeeds/conversion/writer/writer.go index aeae704131c..635ce4e81fe 100644 --- a/vulnfeeds/conversion/writer/writer.go +++ b/vulnfeeds/conversion/writer/writer.go @@ -32,6 +32,7 @@ const ( // hashMetadataKey is the key for the sha256 hash in the GCS object metadata. hashMetadataKey = "sha256-hash" overrideFolder = "osv-output-overrides" // location of overrides within bucket + parallelStartYear = 2018 ) // ErrUploadSkipped indicates that an upload was intentionally skipped because @@ -309,7 +310,12 @@ func UploadVulnsToGCS( func HandleDeletion(ctx context.Context, outBkt *storage.BucketHandle, osvOutputPath string, validVulnIDs []string) { // Check if any need to be deleted - bucketObjects, err := gcs.ListBucketObjects(ctx, outBkt, osvOutputPath) + currentYear := time.Now().Year() + var breakdowns []string + for year := parallelStartYear; year <= currentYear; year++ { + breakdowns = append(breakdowns, fmt.Sprintf("CVE-%d-", year)) + } + bucketObjects, err := gcs.ListObjectsFast(ctx, outBkt, osvOutputPath, breakdowns) if err != nil { logger.Error("Failed to list bucket objects for deletion check, skipping deletion.", slog.Any("err", err)) return diff --git a/vulnfeeds/gcs-tools/gcs.go b/vulnfeeds/gcs-tools/gcs.go index 2b32b241f5f..96488480bf5 100644 --- a/vulnfeeds/gcs-tools/gcs.go +++ b/vulnfeeds/gcs-tools/gcs.go @@ -9,6 +9,7 @@ import ( "log/slog" "os" "path/filepath" + "slices" "strings" "sync" "sync/atomic" @@ -221,12 +222,6 @@ func DownloadBucket(ctx context.Context, bkt *storage.BucketHandle, prefix strin return nil } -// ListBucketObjects lists the names of all objects in a Google Cloud Storage bucket. -// It does not download the file contents. -func ListBucketObjects(ctx context.Context, bucket *storage.BucketHandle, prefix string) ([]string, error) { - return ListBucketObjectsQuery(ctx, bucket, &storage.Query{Prefix: prefix}) -} - // ListBucketObjectsQuery lists the names of all objects in a Google Cloud Storage bucket matching the query. // It optimizes the request by only retrieving the Name attribute. func ListBucketObjectsQuery(ctx context.Context, bucket *storage.BucketHandle, query *storage.Query) ([]string, error) { @@ -251,3 +246,81 @@ func ListBucketObjectsQuery(ctx context.Context, bucket *storage.BucketHandle, q return filenames, nil } + +// ListObjectsFast lists objects in parallel by splitting the prefix space. +// It takes a globalPrefix and a list of breakdownPrefixes. +// It partitions the space using StartOffset and EndOffset based on the breakdowns. +// breakdownPrefixes should be sorted lexicographically. +func ListObjectsFast(ctx context.Context, bucket *storage.BucketHandle, globalPrefix string, breakdownPrefixes []string) ([]string, error) { + globalPrefix = strings.TrimSuffix(globalPrefix, "/") + + if len(breakdownPrefixes) == 0 { + return ListBucketObjectsQuery(ctx, bucket, &storage.Query{Prefix: globalPrefix}) + } + + // Ensure breakdownPrefixes are sorted + sortedBreakdowns := make([]string, len(breakdownPrefixes)) + copy(sortedBreakdowns, breakdownPrefixes) + slices.Sort(sortedBreakdowns) + + var queries []*storage.Query + globalPrefixWithSlash := globalPrefix + if globalPrefix != "" { + globalPrefixWithSlash = globalPrefix + "/" + } + + // 1. First query: everything before the first breakdown + queries = append(queries, &storage.Query{ + Prefix: globalPrefixWithSlash, + EndOffset: globalPrefixWithSlash + sortedBreakdowns[0], + }) + + // 2. Intermediate queries: ranges between breakdowns + for i := 0; i < len(sortedBreakdowns)-1; i++ { + queries = append(queries, &storage.Query{ + Prefix: globalPrefixWithSlash, + StartOffset: globalPrefixWithSlash + sortedBreakdowns[i], + EndOffset: globalPrefixWithSlash + sortedBreakdowns[i+1], + }) + } + + // 3. Last query: everything from the last breakdown onwards + queries = append(queries, &storage.Query{ + Prefix: globalPrefixWithSlash, + StartOffset: globalPrefixWithSlash + sortedBreakdowns[len(sortedBreakdowns)-1], + }) + + objsChan := make(chan []string, len(queries)) + g, ctx := errgroup.WithContext(ctx) + + for _, q := range queries { + g.Go(func() error { + objs, err := ListBucketObjectsQuery(ctx, bucket, q) + if err != nil { + return err + } + objsChan <- objs + return nil + }) + } + + if err := g.Wait(); err != nil { + return nil, err + } + close(objsChan) + + var allObjs []string + for objs := range objsChan { + allObjs = append(allObjs, objs...) + } + + var filtered []string + for _, obj := range allObjs { + if strings.HasSuffix(obj, "/") { + continue + } + filtered = append(filtered, obj) + } + + return filtered, nil +} diff --git a/vulnfeeds/gcs-tools/gcs_test.go b/vulnfeeds/gcs-tools/gcs_test.go index af8a2c7f94f..dca562117ca 100644 --- a/vulnfeeds/gcs-tools/gcs_test.go +++ b/vulnfeeds/gcs-tools/gcs_test.go @@ -5,6 +5,7 @@ import ( "context" "os" "path/filepath" + "slices" "testing" "github.com/fsouza/fake-gcs-server/fakestorage" @@ -208,3 +209,94 @@ func TestDownloadBucket(t *testing.T) { } }) } + +func TestListObjectsFast(t *testing.T) { + objects := []fakestorage.Object{ + { + ObjectAttrs: fakestorage.ObjectAttrs{ + BucketName: "test-bucket", + Name: "cve5/CVE-1999-0001.json", + }, + }, + { + ObjectAttrs: fakestorage.ObjectAttrs{ + BucketName: "test-bucket", + Name: "cve5/CVE-2017-9999.json", + }, + }, + { + ObjectAttrs: fakestorage.ObjectAttrs{ + BucketName: "test-bucket", + Name: "cve5/CVE-2018-0001.json", + }, + }, + { + ObjectAttrs: fakestorage.ObjectAttrs{ + BucketName: "test-bucket", + Name: "cve5/CVE-2018-9999.json", + }, + }, + { + ObjectAttrs: fakestorage.ObjectAttrs{ + BucketName: "test-bucket", + Name: "cve5/CVE-2018/nested.json", // in the gap + }, + }, + { + ObjectAttrs: fakestorage.ObjectAttrs{ + BucketName: "test-bucket", + Name: "cve5/CVE-2019-0001.json", + }, + }, + { + ObjectAttrs: fakestorage.ObjectAttrs{ + BucketName: "test-bucket", + Name: "cve5/CVE-2020-0001.json", + }, + }, + { + ObjectAttrs: fakestorage.ObjectAttrs{ + BucketName: "test-bucket", + Name: "cve5/non-cve.json", + }, + }, + { + ObjectAttrs: fakestorage.ObjectAttrs{ + BucketName: "test-bucket", + Name: "other/file.json", // should not be listed + }, + }, + } + + server := fakestorage.NewServer(objects) + t.Cleanup(server.Stop) + + client := server.Client() + bkt := client.Bucket("test-bucket") + + breakdowns := []string{"CVE-2018-", "CVE-2019-"} + got, err := ListObjectsFast(context.Background(), bkt, "cve5", breakdowns) + if err != nil { + t.Fatalf("ListObjectsFast failed: %v", err) + } + + expected := []string{ + "cve5/CVE-1999-0001.json", + "cve5/CVE-2017-9999.json", + "cve5/CVE-2018-0001.json", + "cve5/CVE-2018-9999.json", + "cve5/CVE-2018/nested.json", + "cve5/CVE-2019-0001.json", + "cve5/CVE-2020-0001.json", + "cve5/non-cve.json", + } + + // Sort both to compare + slices.Sort(got) + slices.Sort(expected) + + if !slices.Equal(got, expected) { + t.Errorf("ListObjectsFast returned unexpected list.\ngot: %v\nwant: %v", got, expected) + } +} + From 9ff62c2a4e0797f19a8571cab3ffc0ac5c73a516 Mon Sep 17 00:00:00 2001 From: Rex P Date: Thu, 16 Jul 2026 15:38:49 +1000 Subject: [PATCH 4/5] lint --- vulnfeeds/conversion/writer/writer.go | 4 ++-- vulnfeeds/gcs-tools/gcs.go | 5 +++-- vulnfeeds/gcs-tools/gcs_test.go | 1 - 3 files changed, 5 insertions(+), 5 deletions(-) diff --git a/vulnfeeds/conversion/writer/writer.go b/vulnfeeds/conversion/writer/writer.go index 635ce4e81fe..a1b835a12cb 100644 --- a/vulnfeeds/conversion/writer/writer.go +++ b/vulnfeeds/conversion/writer/writer.go @@ -30,8 +30,8 @@ import ( const ( // hashMetadataKey is the key for the sha256 hash in the GCS object metadata. - hashMetadataKey = "sha256-hash" - overrideFolder = "osv-output-overrides" // location of overrides within bucket + hashMetadataKey = "sha256-hash" + overrideFolder = "osv-output-overrides" // location of overrides within bucket parallelStartYear = 2018 ) diff --git a/vulnfeeds/gcs-tools/gcs.go b/vulnfeeds/gcs-tools/gcs.go index 96488480bf5..5313d2140ad 100644 --- a/vulnfeeds/gcs-tools/gcs.go +++ b/vulnfeeds/gcs-tools/gcs.go @@ -276,7 +276,7 @@ func ListObjectsFast(ctx context.Context, bucket *storage.BucketHandle, globalPr }) // 2. Intermediate queries: ranges between breakdowns - for i := 0; i < len(sortedBreakdowns)-1; i++ { + for i := range len(sortedBreakdowns) - 1 { queries = append(queries, &storage.Query{ Prefix: globalPrefixWithSlash, StartOffset: globalPrefixWithSlash + sortedBreakdowns[i], @@ -300,6 +300,7 @@ func ListObjectsFast(ctx context.Context, bucket *storage.BucketHandle, globalPr return err } objsChan <- objs + return nil }) } @@ -314,7 +315,7 @@ func ListObjectsFast(ctx context.Context, bucket *storage.BucketHandle, globalPr allObjs = append(allObjs, objs...) } - var filtered []string + filtered := make([]string, 0, len(allObjs)) for _, obj := range allObjs { if strings.HasSuffix(obj, "/") { continue diff --git a/vulnfeeds/gcs-tools/gcs_test.go b/vulnfeeds/gcs-tools/gcs_test.go index dca562117ca..65d0514ad77 100644 --- a/vulnfeeds/gcs-tools/gcs_test.go +++ b/vulnfeeds/gcs-tools/gcs_test.go @@ -299,4 +299,3 @@ func TestListObjectsFast(t *testing.T) { t.Errorf("ListObjectsFast returned unexpected list.\ngot: %v\nwant: %v", got, expected) } } - From cddf27b48b40d134875b968570e68aa9caa01b98 Mon Sep 17 00:00:00 2001 From: Rex P Date: Thu, 16 Jul 2026 15:41:40 +1000 Subject: [PATCH 5/5] more lint --- vulnfeeds/gcs-tools/gcs.go | 2 +- 1 file changed, 1 insertion(+), 1 deletion(-) diff --git a/vulnfeeds/gcs-tools/gcs.go b/vulnfeeds/gcs-tools/gcs.go index 5313d2140ad..a25634a195c 100644 --- a/vulnfeeds/gcs-tools/gcs.go +++ b/vulnfeeds/gcs-tools/gcs.go @@ -263,7 +263,7 @@ func ListObjectsFast(ctx context.Context, bucket *storage.BucketHandle, globalPr copy(sortedBreakdowns, breakdownPrefixes) slices.Sort(sortedBreakdowns) - var queries []*storage.Query + queries := make([]*storage.Query, 0, len(sortedBreakdowns)+1) globalPrefixWithSlash := globalPrefix if globalPrefix != "" { globalPrefixWithSlash = globalPrefix + "/"