From 348f724d81786205df165759b9771d63ba06bdca Mon Sep 17 00:00:00 2001 From: Rex P Date: Wed, 26 Aug 2026 14:05:39 +1000 Subject: [PATCH 01/13] feat(clients): add WriteObjectStream to CloudStorage interface and GCSClient --- go/osv/clients/cloudstorage.go | 4 +++ go/osv/clients/gcs_client.go | 55 ++++++++++++++++++++++++++++++++++ go/testutils/gcs.go | 10 +++++++ 3 files changed, 69 insertions(+) diff --git a/go/osv/clients/cloudstorage.go b/go/osv/clients/cloudstorage.go index 4c2f3244b58..012781988cd 100644 --- a/go/osv/clients/cloudstorage.go +++ b/go/osv/clients/cloudstorage.go @@ -18,6 +18,7 @@ package clients import ( "context" "errors" + "io" "iter" "time" ) @@ -73,6 +74,9 @@ type CloudStorage interface { // WriteObject writes a complete byte slice to a storage object. WriteObject(ctx context.Context, path string, data []byte, opts *WriteOptions) error + // WriteObjectStream streams data from an io.Reader to a storage object. + WriteObjectStream(ctx context.Context, path string, r io.Reader, opts *WriteOptions) error + // Objects returns an iterator over objects that match the prefix. Objects(ctx context.Context, prefix string) iter.Seq2[*Object, error] diff --git a/go/osv/clients/gcs_client.go b/go/osv/clients/gcs_client.go index 127f0d9c5ad..2ea6df1aed4 100644 --- a/go/osv/clients/gcs_client.go +++ b/go/osv/clients/gcs_client.go @@ -115,6 +115,32 @@ func (c *GCSClient) WriteObject(ctx context.Context, path string, data []byte, o return err } +func (c *GCSClient) WriteObjectStream(ctx context.Context, path string, r io.Reader, opts *WriteOptions) error { + var err error + for i := range numRetries { + if i > 0 { + // Exponential backoff: 1s, 2s, 4s + time.Sleep(time.Duration(1<<(i-1)) * time.Second) + } + if seeker, ok := r.(io.Seeker); ok && i > 0 { + if _, seekErr := seeker.Seek(0, io.SeekStart); seekErr != nil { + return seekErr + } + } + err = c.writeObjectStreamOnce(ctx, path, r, opts) + if err == nil { + return nil + } + // Check if error is not transient and should not be retried + var apiErr *googleapi.Error + if !errors.As(err, &apiErr) || (apiErr.Code < 500 && apiErr.Code != 429) { + return err + } + } + + return err +} + func (c *GCSClient) writeObjectOnce(ctx context.Context, path string, data []byte, opts *WriteOptions) error { obj := c.bucket.Object(path) @@ -146,6 +172,35 @@ func (c *GCSClient) writeObjectOnce(ctx context.Context, path string, data []byt return writer.Close() } +func (c *GCSClient) writeObjectStreamOnce(ctx context.Context, path string, r io.Reader, opts *WriteOptions) error { + obj := c.bucket.Object(path) + + if opts != nil && opts.IfGenerationMatches != nil { + conds := storage.Conditions{GenerationMatch: *opts.IfGenerationMatches} + if *opts.IfGenerationMatches == 0 { + conds = storage.Conditions{DoesNotExist: true} + } + obj = obj.If(conds) + } + + writer := obj.NewWriter(ctx) + if opts != nil { + if opts.CustomTime != nil { + writer.CustomTime = *opts.CustomTime + } + if opts.ContentType != "" { + writer.ContentType = opts.ContentType + } + } + + if _, err := io.Copy(writer, r); err != nil { + _ = writer.Close() + return err + } + + return writer.Close() +} + func (c *GCSClient) Close() error { return c.client.Close() } diff --git a/go/testutils/gcs.go b/go/testutils/gcs.go index 585adf9a401..2f763253142 100644 --- a/go/testutils/gcs.go +++ b/go/testutils/gcs.go @@ -3,6 +3,7 @@ package testutils import ( "context" "hash/crc32" + "io" "iter" "slices" "strings" @@ -125,6 +126,15 @@ func (c *MockStorage) WriteObject(_ context.Context, path string, data []byte, o return nil } +func (c *MockStorage) WriteObjectStream(ctx context.Context, path string, r io.Reader, opts *clients.WriteOptions) error { + data, err := io.ReadAll(r) + if err != nil { + return err + } + + return c.WriteObject(ctx, path, data, opts) +} + func (c *MockStorage) Objects(_ context.Context, prefix string) iter.Seq2[*clients.Object, error] { // Create a snapshot of the keys to iterate over, so we don't hold the lock. c.mu.RLock() From b399b14c8f4f1337976edebdce32b7905bc8367d Mon Sep 17 00:00:00 2001 From: Rex P Date: Wed, 26 Aug 2026 14:17:18 +1000 Subject: [PATCH 02/13] perf(exporter): stream zip/csv generation and use disk-backed scratch staging --- go/cmd/exporter/exporter.go | 81 +++++++++-- go/cmd/exporter/exporter_test.go | 214 ++++++++++++++++++++++++++++ go/cmd/exporter/worker.go | 233 ++++++++++++++++++------------- go/cmd/exporter/writer.go | 135 ++++++++++++++---- 4 files changed, 533 insertions(+), 130 deletions(-) create mode 100644 go/cmd/exporter/exporter_test.go diff --git a/go/cmd/exporter/exporter.go b/go/cmd/exporter/exporter.go index a76c1636a68..e617a75e6f7 100644 --- a/go/cmd/exporter/exporter.go +++ b/go/cmd/exporter/exporter.go @@ -1,5 +1,3 @@ -// Package main runs the exporter, exporting the whole OSV database to the GCS bucket. -// See the README.md for more details. package main import ( @@ -8,6 +6,7 @@ import ( "log/slog" "os" "os/signal" + "path/filepath" "strings" "sync" "syscall" @@ -35,20 +34,38 @@ func main() { ctx, span := otel.Tracer("exporter").Start(ctx, "exporter") defer span.End() + defaultScratchDir := os.Getenv("SCRATCH_DIR") + if defaultScratchDir == "" { + defaultScratchDir = filepath.Join(os.TempDir(), "osv-exporter-scratch") + } + outBucketName := flag.String("bucket", "osv-test-vulnerabilities", "Output bucket or directory name. If -local is true, this is a local path; otherwise, it's a GCS bucket name.") vulnBucketName := flag.String("osv-vulns-bucket", os.Getenv("OSV_VULNERABILITIES_BUCKET"), "GCS bucket to read vulnerability protobufs from. Can also be set with the OSV_VULNERABILITIES_BUCKET environment variable.") uploadToGCS := flag.Bool("upload-to-gcs", false, "If false, writes the output to a local directory specified by -bucket instead of a GCS bucket.") numWorkers := flag.Int("workers", 1000, "The total number of concurrent workers to use for downloading from GCS and writing the output.") breakdownPrefixesStr := flag.String("breakdown-prefixes", "", "Comma-separated list of prefix breakdowns for parallel GCS object listing.") + scratchDirFlag := flag.String("scratch-dir", defaultScratchDir, "Directory to stage temporary JSON and zip files.") flag.Parse() + scratchDir := *scratchDirFlag + if err := os.MkdirAll(scratchDir, 0755); err != nil { + logger.FatalContext(ctx, "failed to create scratch directory", slog.String("dir", scratchDir), slog.Any("err", err)) + } + stagingDir, err := os.MkdirTemp(scratchDir, "osv-exporter-staging-*") + if err != nil { + logger.FatalContext(ctx, "failed to create staging directory in scratch dir", slog.String("dir", scratchDir), slog.Any("err", err)) + } + defer os.RemoveAll(stagingDir) + logger.InfoContext(ctx, "exporter starting", slog.String("bucket", *outBucketName), slog.String("osv-vulns-bucket", *vulnBucketName), slog.Bool("upload-to-gcs", *uploadToGCS), slog.Int("workers", *numWorkers), - slog.String("breakdown-prefixes", *breakdownPrefixesStr)) + slog.String("breakdown-prefixes", *breakdownPrefixesStr), + slog.String("scratch-dir", scratchDir), + slog.String("staging-dir", stagingDir)) if *vulnBucketName == "" { logger.FatalContext(ctx, "OSV_VULNERABILITIES_BUCKET must be set") @@ -102,7 +119,7 @@ func main() { } var routerWg sync.WaitGroup routerWg.Add(1) - go ecosystemRouter(ctx, downloaderToRouterCh, routerToWriteCh, &routerWg) + go ecosystemRouter(ctx, downloaderToRouterCh, routerToWriteCh, stagingDir, &routerWg) MainLoop: for objName, err := range vulnClient.ObjectsFast(ctx, gcsProtoPrefix, breakdownPrefixes) { @@ -132,14 +149,20 @@ MainLoop: // ecosystemRouter receives vulnerabilities from inCh and fans them out to the // appropriate ecosystemWorker. It creates workers on-demand for each new // ecosystem encountered. It also sends every vulnerability to the allEcosystemWorker. -func ecosystemRouter(ctx context.Context, inCh <-chan *osvschema.Vulnerability, outCh chan<- writeMsg, wg *sync.WaitGroup) { +func ecosystemRouter(ctx context.Context, inCh <-chan *osvschema.Vulnerability, outCh chan<- writeMsg, scratchDir string, wg *sync.WaitGroup) { defer wg.Done() logger.InfoContext(ctx, "ecosystem router starting") workers := make(map[string]*ecosystemWorker) var workersWg sync.WaitGroup vulnCounter := 0 + var vanirVulns []vulnMeta + + vulnsDir := filepath.Join(scratchDir, "vulns") + if err := os.MkdirAll(vulnsDir, 0755); err != nil { + logger.FatalContext(ctx, "failed to create scratch vulns directory", slog.Any("err", err)) + } - allEcosystemWorker := newAllEcosystemWorker(ctx, outCh, &workersWg) + allEcosystemWorker := newAllEcosystemWorker(ctx, scratchDir, outCh, &workersWg) RouterLoop: for { @@ -154,6 +177,36 @@ RouterLoop: } } vulnCounter++ + + // Marshal JSON ONCE for this vulnerability. + b, err := marshalToJSON(vuln) + if err != nil { + logger.ErrorContext(ctx, "failed to marshal vulnerability to json", slog.String("id", vuln.GetId()), slog.Any("err", err)) + continue + } + + // Cache to local scratch disk for later ZIP generation. + localPath := filepath.Join(vulnsDir, vuln.GetId()+".json") + if err := os.WriteFile(localPath, b, 0600); err != nil { + logger.ErrorContext(ctx, "failed to write cached vulnerability to disk", slog.String("id", vuln.GetId()), slog.Any("err", err)) + continue + } + + meta := vulnMeta{ + id: vuln.GetId(), + modified: vuln.GetModified().AsTime(), + localPath: localPath, + } + + // Check for Vanir signatures + for _, aff := range vuln.GetAffected() { + spec := aff.GetDatabaseSpecific() + if _, ok := spec.GetFields()["vanir_signatures"]; ok { + vanirVulns = append(vanirVulns, meta) + break + } + } + ecosystems := make(map[string]struct{}) for _, aff := range vuln.GetAffected() { eco := aff.GetPackage().GetEcosystem() @@ -175,17 +228,22 @@ RouterLoop: ecoNames = append(ecoNames, eco) worker, ok := workers[eco] if !ok { - worker = newEcosystemWorker(ctx, eco, outCh, &workersWg) + worker = newEcosystemWorker(ctx, eco, scratchDir, outCh, &workersWg) workers[eco] = worker } select { - case worker.inCh <- vuln: + case worker.inCh <- meta: + case <-ctx.Done(): + break RouterLoop + } + select { + case outCh <- writeMsg{path: filepath.Join(eco, vuln.GetId()) + ".json", mimeType: "application/json", data: b}: case <-ctx.Done(): break RouterLoop } } select { - case allEcosystemWorker.inCh <- vulnAndEcos{Vulnerability: vuln, ecosystems: ecoNames}: + case allEcosystemWorker.inCh <- vulnAndEcos{meta: meta, ecosystems: ecoNames}: case <-ctx.Done(): break RouterLoop } @@ -196,6 +254,11 @@ RouterLoop: } allEcosystemWorker.Finish() workersWg.Wait() + + if len(vanirVulns) > 0 && ctx.Err() == nil { + writeVanir(ctx, vanirVulns, outCh, scratchDir) + } + if ctx.Err() == nil { logger.InfoContext(ctx, "ecosystem router finished, all vulnerabilities dispatched", slog.Int("total_vulnerabilities", vulnCounter)) } else { diff --git a/go/cmd/exporter/exporter_test.go b/go/cmd/exporter/exporter_test.go new file mode 100644 index 00000000000..db6d5a4a350 --- /dev/null +++ b/go/cmd/exporter/exporter_test.go @@ -0,0 +1,214 @@ +package main + +import ( + "archive/zip" + "bytes" + "context" + "encoding/csv" + "encoding/json" + "os" + "path/filepath" + "sync" + "testing" + "time" + + "github.com/google/osv.dev/go/testutils" + "github.com/ossf/osv-schema/bindings/go/osvschema" + "google.golang.org/protobuf/types/known/structpb" + "google.golang.org/protobuf/types/known/timestamppb" +) + +func TestWriter_GCS_StreamUpload(t *testing.T) { + storage := testutils.NewMockStorage() + tmpDir := t.TempDir() + tmpFile := filepath.Join(tmpDir, "sample.zip") + content := []byte("zip binary contents") + if err := os.WriteFile(tmpFile, content, 0600); err != nil { + t.Fatalf("failed to write temp file: %v", err) + } + + runWriter(t, storage, []writeMsg{ + {path: "all.zip", mimeType: "application/zip", filePath: tmpFile}, + }) + + objPath := filepath.Join("out", "all.zip") + got, err := storage.ReadObject(t.Context(), objPath) + if err != nil { + t.Fatalf("ReadObject(%s) failed: %v", objPath, err) + } + if !bytes.Equal(got, content) { + t.Errorf("expected %q, got %q", content, got) + } +} + +func TestWriter_GCS_StreamSkipsUnchanged(t *testing.T) { + storage := testutils.NewMockStorage() + tmpDir := t.TempDir() + tmpFile := filepath.Join(tmpDir, "sample.zip") + content := []byte("zip binary contents") + if err := os.WriteFile(tmpFile, content, 0600); err != nil { + t.Fatalf("failed to write temp file: %v", err) + } + + objPath := filepath.Join("out", "all.zip") + if err := storage.WriteObject(t.Context(), objPath, content, nil); err != nil { + t.Fatalf("setup WriteObject failed: %v", err) + } + attrsBefore, _ := storage.ReadObjectAttrs(t.Context(), objPath) + + runWriter(t, storage, []writeMsg{ + {path: "all.zip", mimeType: "application/zip", filePath: tmpFile}, + }) + + attrsAfter, err := storage.ReadObjectAttrs(t.Context(), objPath) + if err != nil { + t.Fatalf("ReadObjectAttrs failed: %v", err) + } + if attrsAfter.Generation != attrsBefore.Generation { + t.Errorf("expected generation %d (skipped), got %d", attrsBefore.Generation, attrsAfter.Generation) + } +} + +func TestExporterPipeline_EndToEnd(t *testing.T) { + ctx, cancel := context.WithCancel(t.Context()) + defer cancel() + + scratchDir := t.TempDir() + storage := testutils.NewMockStorage() + + inCh := make(chan *osvschema.Vulnerability, 10) + routerToWriteCh := make(chan writeMsg, 100) + + var writerWg sync.WaitGroup + writerWg.Add(1) + go writer(ctx, cancel, routerToWriteCh, storage, "export", &writerWg) + + var routerWg sync.WaitGroup + routerWg.Add(1) + go ecosystemRouter(ctx, inCh, routerToWriteCh, scratchDir, &routerWg) + + time1 := time.Date(2023, 1, 1, 12, 0, 0, 0, time.UTC) + time2 := time.Date(2023, 2, 1, 12, 0, 0, 0, time.UTC) + + // Create test vulnerability 1: PyPI + vuln1 := &osvschema.Vulnerability{ + Id: "GHSA-pypi-1", + Modified: timestamppb.New(time1), + Affected: []*osvschema.Affected{ + { + Package: &osvschema.Package{ + Ecosystem: "PyPI", + Name: "requests", + }, + }, + }, + } + + // Create test vulnerability 2: npm and GIT with Vanir signatures + vanirField, _ := structpb.NewValue("test-signature") + vuln2 := &osvschema.Vulnerability{ + Id: "GHSA-npm-git-2", + Modified: timestamppb.New(time2), + Affected: []*osvschema.Affected{ + { + Package: &osvschema.Package{ + Ecosystem: "npm", + Name: "lodash", + }, + Ranges: []*osvschema.Range{ + { + Type: osvschema.Range_GIT, + }, + }, + DatabaseSpecific: &structpb.Struct{ + Fields: map[string]*structpb.Value{ + "vanir_signatures": vanirField, + }, + }, + }, + }, + } + + inCh <- vuln1 + inCh <- vuln2 + close(inCh) + + routerWg.Wait() + close(routerToWriteCh) + writerWg.Wait() + + // 1. Verify individual JSON outputs + pypiJSON, err := storage.ReadObject(ctx, "export/PyPI/GHSA-pypi-1.json") + if err != nil { + t.Fatalf("expected PyPI/GHSA-pypi-1.json in storage: %v", err) + } + if !bytes.Contains(pypiJSON, []byte(`"id":"GHSA-pypi-1"`)) { + t.Errorf("PyPI JSON content mismatch: %s", string(pypiJSON)) + } + + npmJSON, err := storage.ReadObject(ctx, "export/npm/GHSA-npm-git-2.json") + if err != nil { + t.Fatalf("expected npm/GHSA-npm-git-2.json in storage: %v", err) + } + if !bytes.Contains(npmJSON, []byte(`"id":"GHSA-npm-git-2"`)) { + t.Errorf("npm JSON content mismatch: %s", string(npmJSON)) + } + + // 2. Verify all.zip contains all JSON files + allZipBytes, err := storage.ReadObject(ctx, "export/all.zip") + if err != nil { + t.Fatalf("expected all.zip: %v", err) + } + zipReader, err := zip.NewReader(bytes.NewReader(allZipBytes), int64(len(allZipBytes))) + if err != nil { + t.Fatalf("failed to open all.zip: %v", err) + } + var zipNames []string + for _, f := range zipReader.File { + zipNames = append(zipNames, f.Name) + } + if len(zipNames) != 2 { + t.Errorf("expected 2 files in all.zip, got %v", zipNames) + } + + // 3. Verify modified_id.csv ordering (descending by modified time) + csvBytes, err := storage.ReadObject(ctx, "export/modified_id.csv") + if err != nil { + t.Fatalf("expected modified_id.csv: %v", err) + } + csvReader := csv.NewReader(bytes.NewReader(csvBytes)) + records, err := csvReader.ReadAll() + if err != nil { + t.Fatalf("failed to parse modified_id.csv: %v", err) + } + // vuln2 has later modified time, so its ecosystems (GIT, npm) should appear before PyPI + if len(records) < 3 { + t.Fatalf("expected at least 3 records in modified_id.csv, got %d", len(records)) + } + if records[0][0] != time2.Format(time.RFC3339Nano) { + t.Errorf("expected first record to be time2, got %s", records[0][0]) + } + + // 4. Verify ecosystems.txt + ecoTxtBytes, err := storage.ReadObject(ctx, "export/ecosystems.txt") + if err != nil { + t.Fatalf("expected ecosystems.txt: %v", err) + } + expectedEco := "GIT\nPyPI\nnpm\n" + if string(ecoTxtBytes) != expectedEco { + t.Errorf("expected %q in ecosystems.txt, got %q", expectedEco, string(ecoTxtBytes)) + } + + // 5. Verify Vanir signatures file (GIT/osv_git.json) + vanirBytes, err := storage.ReadObject(ctx, "export/GIT/osv_git.json") + if err != nil { + t.Fatalf("expected GIT/osv_git.json: %v", err) + } + var vanirList []map[string]any + if err := json.Unmarshal(vanirBytes, &vanirList); err != nil { + t.Fatalf("failed to unmarshal vanir JSON: %v", err) + } + if len(vanirList) != 1 || vanirList[0]["id"] != "GHSA-npm-git-2" { + t.Errorf("unexpected vanir content: %s", string(vanirBytes)) + } +} diff --git a/go/cmd/exporter/worker.go b/go/cmd/exporter/worker.go index 4029ff2cf02..4d0e2400a02 100644 --- a/go/cmd/exporter/worker.go +++ b/go/cmd/exporter/worker.go @@ -10,6 +10,7 @@ import ( "io" "log/slog" "maps" + "os" "path/filepath" "slices" "strings" @@ -30,18 +31,33 @@ const ( ecosystemsFilename = "ecosystems.txt" ) +// vulnMeta holds the ID, modified time, and local staging file path for a vulnerability. +type vulnMeta struct { + id string + modified time.Time + localPath string +} + +// csvEntry holds the unix timestamp in nanoseconds and the relative entry path. +type csvEntry struct { + modified int64 + path string +} + // ecosystemWorker processes vulnerabilities for a single ecosystem. type ecosystemWorker struct { - ecosystem string - inCh chan *osvschema.Vulnerability + ecosystem string + scratchDir string + inCh chan vulnMeta } // newEcosystemWorker creates and starts a new ecosystemWorker. -func newEcosystemWorker(ctx context.Context, ecosystem string, outCh chan<- writeMsg, wg *sync.WaitGroup) *ecosystemWorker { - ch := make(chan *osvschema.Vulnerability, 100) +func newEcosystemWorker(ctx context.Context, ecosystem string, scratchDir string, outCh chan<- writeMsg, wg *sync.WaitGroup) *ecosystemWorker { + ch := make(chan vulnMeta, 100) worker := &ecosystemWorker{ - ecosystem: ecosystem, - inCh: ch, + ecosystem: ecosystem, + scratchDir: scratchDir, + inCh: ch, } wg.Add(1) go worker.run(ctx, outCh, wg) @@ -49,16 +65,9 @@ func newEcosystemWorker(ctx context.Context, ecosystem string, outCh chan<- writ return worker } -// vulnData holds the ID and marshalled JSON data for a vulnerability. -type vulnData struct { - id string - modified time.Time - data []byte -} - -// run is the main loop for the ecosystemWorker. It receives vulnerabilities, +// run is the main loop for the ecosystemWorker. It receives vulnerability metadata, // aggregates them, and upon completion, writes out the ecosystem-specific -// zip, csv, and (for GIT) vanir files. +// zip and csv files. func (w *ecosystemWorker) run(ctx context.Context, outCh chan<- writeMsg, wg *sync.WaitGroup) { defer wg.Done() ctx, span := otel.Tracer("exporter").Start(ctx, w.ecosystem) @@ -66,39 +75,11 @@ func (w *ecosystemWorker) run(ctx context.Context, outCh chan<- writeMsg, wg *sy logger.InfoContext(ctx, "new ecosystem worker started", slog.String("ecosystem", w.ecosystem)) // 500 size is around the minimum each ecosystem would need, most ecosystems are much bigger. - allVulns := make([]vulnData, 0, 500) - csvData := make([][]string, 0, 500) - var vanirVulns []vulnData + allVulns := make([]vulnMeta, 0, 500) + csvData := make([]csvEntry, 0, 500) for v := range w.inCh { - // Process vulnerability. - b, err := marshalToJSON(v) - if err != nil { - logger.ErrorContext(ctx, "failed to marshal vulnerability to json", slog.String("id", v.GetId()), slog.Any("err", err)) - continue - } - - // Wait to send the result, or be cancelled. - select { - case outCh <- writeMsg{path: filepath.Join(w.ecosystem, v.GetId()) + ".json", mimeType: "application/json", data: b}: - case <-ctx.Done(): - logger.WarnContext(ctx, "ecosystem worker cancelled", slog.String("ecosystem", w.ecosystem), slog.Any("err", ctx.Err())) - return - } - - modified := v.GetModified().AsTime() - allVulns = append(allVulns, vulnData{id: v.GetId(), modified: modified, data: b}) - csvData = append(csvData, []string{modified.Format(time.RFC3339Nano), v.GetId()}) - - // For GIT ecosystem, we want to make a file containing every vulnerability with vanir signatures - if w.ecosystem == gitEcosystem { - for _, aff := range v.GetAffected() { - spec := aff.GetDatabaseSpecific() - if _, ok := spec.GetFields()["vanir_signatures"]; ok { - vanirVulns = append(vanirVulns, vulnData{id: v.GetId(), data: b}) - break - } - } - } + allVulns = append(allVulns, v) + csvData = append(csvData, csvEntry{modified: v.modified.UnixNano(), path: v.id}) } if ctx.Err() != nil { @@ -106,11 +87,8 @@ func (w *ecosystemWorker) run(ctx context.Context, outCh chan<- writeMsg, wg *sy } logger.InfoContext(ctx, "All vulnerabilities processed", slog.String("ecosystem", w.ecosystem)) - writeModifiedIDCSV(ctx, filepath.Join(w.ecosystem, modifiedCSVFilename), csvData, outCh) - writeZIP(ctx, filepath.Join(w.ecosystem, allZipFilename), allVulns, outCh) - if w.ecosystem == gitEcosystem { - writeVanir(ctx, vanirVulns, outCh) - } + writeModifiedIDCSV(ctx, filepath.Join(w.ecosystem, modifiedCSVFilename), csvData, outCh, w.scratchDir) + writeZIP(ctx, filepath.Join(w.ecosystem, allZipFilename), allVulns, outCh, w.scratchDir) logger.InfoContext(ctx, "ecosystem worker finished processing", slog.String("ecosystem", w.ecosystem)) } @@ -119,24 +97,25 @@ func (w *ecosystemWorker) Finish() { close(w.inCh) } -// vulnAndEcos holds a vulnerability and the list of ecosystems it belongs to. +// vulnAndEcos holds a vulnerability metadata and the list of ecosystems it belongs to. type vulnAndEcos struct { - *osvschema.Vulnerability - + meta vulnMeta ecosystems []string } // allEcosystemWorker processes all vulnerabilities from all ecosystems to create // the global export files. type allEcosystemWorker struct { - inCh chan vulnAndEcos + scratchDir string + inCh chan vulnAndEcos } // newAllEcosystemWorker creates and starts a new allEcosystemWorker. -func newAllEcosystemWorker(ctx context.Context, outCh chan<- writeMsg, wg *sync.WaitGroup) *allEcosystemWorker { +func newAllEcosystemWorker(ctx context.Context, scratchDir string, outCh chan<- writeMsg, wg *sync.WaitGroup) *allEcosystemWorker { ch := make(chan vulnAndEcos, 100) worker := &allEcosystemWorker{ - inCh: ch, + scratchDir: scratchDir, + inCh: ch, } wg.Add(1) go worker.run(ctx, outCh, wg) @@ -153,21 +132,15 @@ func (w *allEcosystemWorker) run(ctx context.Context, outCh chan<- writeMsg, wg logger.InfoContext(ctx, "all-ecosystem worker started") // We have currently about 1.8 million entries, so start at 100k - allVulns := make([]vulnData, 0, 100000) - csvData := make([][]string, 0, 100000) + allVulns := make([]vulnMeta, 0, 100000) + csvData := make([]csvEntry, 0, 100000) ecosystems := make(map[string]struct{}) for v := range w.inCh { - b, err := marshalToJSON(v.Vulnerability) - if err != nil { - logger.ErrorContext(ctx, "failed to marshal vulnerability to json", slog.String("id", v.GetId()), slog.Any("err", err)) - continue - } - modified := v.GetModified().AsTime() - allVulns = append(allVulns, vulnData{id: v.GetId(), modified: modified, data: b}) + allVulns = append(allVulns, v.meta) for _, e := range v.ecosystems { ecosystems[e] = struct{}{} - csvData = append(csvData, []string{modified.Format(time.RFC3339Nano), e + "/" + v.GetId()}) - if len(csvData)%10000 == 0 { + csvData = append(csvData, csvEntry{modified: v.meta.modified.UnixNano(), path: e + "/" + v.meta.id}) + if len(csvData)%50000 == 0 { logger.InfoContext(ctx, "processed N vulnerabilities", slog.Int("n", len(csvData))) } } @@ -177,8 +150,8 @@ func (w *allEcosystemWorker) run(ctx context.Context, outCh chan<- writeMsg, wg return } - writeModifiedIDCSV(ctx, modifiedCSVFilename, csvData, outCh) - writeZIP(ctx, allZipFilename, allVulns, outCh) + writeModifiedIDCSV(ctx, modifiedCSVFilename, csvData, outCh, w.scratchDir) + writeZIP(ctx, allZipFilename, allVulns, outCh, w.scratchDir) ecos := slices.Collect(maps.Keys(ecosystems)) slices.Sort(ecos) ecoString := strings.Join(ecos, "\n") + "\n" @@ -214,35 +187,64 @@ func write(ctx context.Context, path string, data []byte, mimeType string, outCh } } -// writeModifiedIDCSV constructs and writes a modified_id.csv file. -func writeModifiedIDCSV(ctx context.Context, path string, csvData [][]string, outCh chan<- writeMsg) { +// writeStream is a helper to send a streaming file writeMsg to the writer channel. +func writeStream(ctx context.Context, path string, filePath string, mimeType string, outCh chan<- writeMsg) { + select { + case outCh <- writeMsg{path: path, mimeType: mimeType, filePath: filePath}: + case <-ctx.Done(): + } +} + +// writeModifiedIDCSV constructs and writes a modified_id.csv file by streaming to a temporary file. +func writeModifiedIDCSV(ctx context.Context, path string, csvData []csvEntry, outCh chan<- writeMsg, scratchDir string) { logger.InfoContext(ctx, "constructing csv file", slog.String("path", path)) - slices.SortFunc(csvData, func(a, b []string) int { + slices.SortFunc(csvData, func(a, b csvEntry) int { return cmp.Or( - -cmp.Compare(a[0], b[0]), // Modified date, descending - cmp.Compare(a[1], b[1]), // path/vuln ID, ascending + -cmp.Compare(a.modified, b.modified), // Modified date, descending + cmp.Compare(a.path, b.path), // path/vuln ID, ascending ) }) - var buf bytes.Buffer - wr := csv.NewWriter(&buf) - if err := wr.WriteAll(csvData); err != nil { - logger.ErrorContext(ctx, "failed writing csv", slog.String("path", path), slog.Any("err", err)) + tmpCsv, err := os.CreateTemp(scratchDir, "csv-*.tmp") + if err != nil { + logger.ErrorContext(ctx, "failed to create temp csv file", slog.String("path", path), slog.Any("err", err)) return } + defer tmpCsv.Close() + + wr := csv.NewWriter(tmpCsv) + for _, entry := range csvData { + t := time.Unix(0, entry.modified).UTC().Format(time.RFC3339Nano) + if err := wr.Write([]string{t, entry.path}); err != nil { + logger.ErrorContext(ctx, "failed writing csv line", slog.String("path", path), slog.Any("err", err)) + return + } + } wr.Flush() + if err := wr.Error(); err != nil { + logger.ErrorContext(ctx, "failed flushing csv", slog.String("path", path), slog.Any("err", err)) + return + } + logger.InfoContext(ctx, "writing csv file", slog.String("path", path)) - write(ctx, path, buf.Bytes(), "text/csv", outCh) + writeStream(ctx, path, tmpCsv.Name(), "text/csv", outCh) } -// writeZIP constructs and writes an all.zip file. -func writeZIP(ctx context.Context, path string, allVulns []vulnData, outCh chan<- writeMsg) { +// writeZIP constructs and writes a zip file by streaming from local files to a temporary zip file. +func writeZIP(ctx context.Context, path string, allVulns []vulnMeta, outCh chan<- writeMsg, scratchDir string) { logger.InfoContext(ctx, "constructing zip file", slog.String("path", path)) - slices.SortFunc(allVulns, func(a, b vulnData) int { + slices.SortFunc(allVulns, func(a, b vulnMeta) int { return cmp.Compare(a.id, b.id) }) - var buf bytes.Buffer - wr := zip.NewWriter(&buf) + + tmpZip, err := os.CreateTemp(scratchDir, "zip-*.tmp") + if err != nil { + logger.ErrorContext(ctx, "failed to create temp zip file", slog.String("path", path), slog.Any("err", err)) + return + } + defer tmpZip.Close() + + wr := zip.NewWriter(tmpZip) for _, vuln := range allVulns { w, err := wr.CreateHeader(&zip.FileHeader{ Name: vuln.id + ".json", @@ -253,29 +255,66 @@ func writeZIP(ctx context.Context, path string, allVulns []vulnData, outCh chan< logger.ErrorContext(ctx, "failed to create vuln json in zip file", slog.String("id", vuln.id), slog.Any("err", err)) continue } - r := bytes.NewReader(vuln.data) - if _, err := io.Copy(w, r); err != nil { + f, err := os.Open(vuln.localPath) + if err != nil { + logger.ErrorContext(ctx, "failed to open local vuln json file", slog.String("path", vuln.localPath), slog.Any("err", err)) + continue + } + if _, err := io.Copy(w, f); err != nil { logger.ErrorContext(ctx, "failed to write vuln json in zip file", slog.String("id", vuln.id), slog.Any("err", err)) } + f.Close() } if err := wr.Close(); err != nil { logger.ErrorContext(ctx, "failed to close zip writer", slog.String("path", path), slog.Any("err", err)) + return } + logger.InfoContext(ctx, "writing zip file", slog.String("path", path)) - write(ctx, path, buf.Bytes(), "application/zip", outCh) + writeStream(ctx, path, tmpZip.Name(), "application/zip", outCh) } // writeVanir constructs and writes the osv_git.json file containing vulnerabilities with Vanir signatures. -func writeVanir(ctx context.Context, vanirVulns []vulnData, outCh chan<- writeMsg) { - slices.SortFunc(vanirVulns, func(a, b vulnData) int { return cmp.Compare(a.id, b.id) }) - vulns := make([]json.RawMessage, len(vanirVulns)) +func writeVanir(ctx context.Context, vanirVulns []vulnMeta, outCh chan<- writeMsg, scratchDir string) { + slices.SortFunc(vanirVulns, func(a, b vulnMeta) int { return cmp.Compare(a.id, b.id) }) + + tmpVanir, err := os.CreateTemp(scratchDir, "vanir-*.json") + if err != nil { + logger.ErrorContext(ctx, "failed to create temp vanir file", slog.Any("err", err)) + return + } + defer tmpVanir.Close() + + if _, err := tmpVanir.WriteString("[\n"); err != nil { + logger.ErrorContext(ctx, "failed to write vanir header", slog.Any("err", err)) + return + } for i, v := range vanirVulns { - vulns[i] = v.data + data, err := os.ReadFile(v.localPath) + if err != nil { + logger.ErrorContext(ctx, "failed to read local vanir file", slog.String("path", v.localPath), slog.Any("err", err)) + continue + } + if _, err := tmpVanir.Write(data); err != nil { + logger.ErrorContext(ctx, "failed to write vanir entry", slog.Any("err", err)) + return + } + if i < len(vanirVulns)-1 { + if _, err := tmpVanir.WriteString(",\n"); err != nil { + logger.ErrorContext(ctx, "failed to write vanir separator", slog.Any("err", err)) + return + } + } else { + if _, err := tmpVanir.WriteString("\n"); err != nil { + logger.ErrorContext(ctx, "failed to write vanir newline", slog.Any("err", err)) + return + } + } } - finalJSON, err := json.Marshal(vulns) - if err != nil { - logger.ErrorContext(ctx, "failed to marshal vanir JSON file", slog.Any("err", err)) + if _, err := tmpVanir.WriteString("]\n"); err != nil { + logger.ErrorContext(ctx, "failed to write vanir footer", slog.Any("err", err)) return } - write(ctx, filepath.Join(gitEcosystem, vanirVulnsFilename), finalJSON, "application/json", outCh) + + writeStream(ctx, filepath.Join(gitEcosystem, vanirVulnsFilename), tmpVanir.Name(), "application/json", outCh) } diff --git a/go/cmd/exporter/writer.go b/go/cmd/exporter/writer.go index 29533fa7ba5..a10d550e03d 100644 --- a/go/cmd/exporter/writer.go +++ b/go/cmd/exporter/writer.go @@ -4,6 +4,7 @@ import ( "context" "errors" "hash/crc32" + "io" "log/slog" "os" "path/filepath" @@ -21,6 +22,7 @@ type writeMsg struct { path string mimeType string data []byte + filePath string // If set, stream from local file instead of holding data in memory } // writer is a worker that receives writeMsgs and writes them to either a GCS @@ -29,39 +31,124 @@ func writer(ctx context.Context, cancel context.CancelFunc, inCh <-chan writeMsg defer wg.Done() for msg := range inCh { path := filepath.Join(pathPrefix, msg.path) - if client != nil { - // Skip the upload if the object already has the same content. - if gcsContentUnchanged(ctx, client, path, msg.data) { - continue - } - err := client.WriteObject(ctx, path, msg.data, &clients.WriteOptions{ - ContentType: msg.mimeType, - }) - if err != nil { - logger.Error("failed to write file", slog.String("path", path), slog.Any("err", err)) - cancel() - - break + if msg.filePath != "" { + if client != nil { + if gcsFileUnchanged(ctx, client, path, msg.filePath) { + continue + } + f, err := os.Open(msg.filePath) + if err != nil { + logger.Error("failed to open local file for upload", slog.String("path", path), slog.String("file", msg.filePath), slog.Any("err", err)) + cancel() + + break + } + err = client.WriteObjectStream(ctx, path, f, &clients.WriteOptions{ + ContentType: msg.mimeType, + }) + f.Close() + if err != nil { + logger.Error("failed to stream write file", slog.String("path", path), slog.Any("err", err)) + cancel() + + break + } + } else { + // Write locally: copy from msg.filePath to path + dir := filepath.Dir(path) + if err := os.MkdirAll(dir, 0755); err != nil { + logger.Error("failed to create directories", slog.String("dir", dir), slog.Any("err", err)) + cancel() + + break + } + if err := copyFile(msg.filePath, path); err != nil { + logger.Error("failed to copy file locally", slog.String("src", msg.filePath), slog.String("dst", path), slog.Any("err", err)) + cancel() + + break + } } } else { - // Write locally. - dir := filepath.Dir(path) - if err := os.MkdirAll(dir, 0755); err != nil { - logger.Error("failed to create directories", slog.String("dir", dir), slog.Any("err", err)) - cancel() + if client != nil { + // Skip the upload if the object already has the same content. + if gcsContentUnchanged(ctx, client, path, msg.data) { + continue + } + err := client.WriteObject(ctx, path, msg.data, &clients.WriteOptions{ + ContentType: msg.mimeType, + }) + if err != nil { + logger.Error("failed to write file", slog.String("path", path), slog.Any("err", err)) + cancel() - break - } - if err := os.WriteFile(path, msg.data, 0600); err != nil { - logger.Error("failed to write file", slog.String("path", path), slog.Any("err", err)) - cancel() + break + } + } else { + // Write locally. + dir := filepath.Dir(path) + if err := os.MkdirAll(dir, 0755); err != nil { + logger.Error("failed to create directories", slog.String("dir", dir), slog.Any("err", err)) + cancel() + + break + } + if err := os.WriteFile(path, msg.data, 0600); err != nil { + logger.Error("failed to write file", slog.String("path", path), slog.Any("err", err)) + cancel() - break + break + } } } } } +func copyFile(src, dst string) error { + in, err := os.Open(src) + if err != nil { + return err + } + defer in.Close() + + out, err := os.OpenFile(dst, os.O_WRONLY|os.O_CREATE|os.O_TRUNC, 0600) + if err != nil { + return err + } + defer out.Close() + + _, err = io.Copy(out, in) + + return err +} + +func gcsFileUnchanged(ctx context.Context, client clients.CloudStorage, path string, filePath string) bool { + attrs, err := client.ReadObjectAttrs(ctx, path) + if err != nil { + if !errors.Is(err, clients.ErrNotFound) { + logger.WarnContext(ctx, "failed to read object attrs, proceeding with upload", slog.String("path", path), slog.Any("err", err)) + } + + return false + } + f, err := os.Open(filePath) + if err != nil { + return false + } + defer f.Close() + h := crc32.New(crc32cTable) + if _, err := io.Copy(h, f); err != nil { + return false + } + if attrs.CRC32C == h.Sum32() { + logger.InfoContext(ctx, "skipping upload, content unchanged", slog.String("path", path)) + + return true + } + + return false +} + // gcsContentUnchanged returns true if the object at path already has the same // CRC32C checksum as data, meaning the upload would be a no-op. Any error // reading the object's attributes (other than ErrNotFound) is logged and From ff7b7d3b749b8a4e9ca5a1d855616344d40a8ee0 Mon Sep 17 00:00:00 2001 From: Rex P Date: Wed, 26 Aug 2026 14:18:03 +1000 Subject: [PATCH 03/13] perf(deploy): rightsize exporter resources, allocate scratch volume, and move to default pool --- .../gke-workers/base/core/exporter.yaml | 25 +++++++++++-------- 1 file changed, 15 insertions(+), 10 deletions(-) diff --git a/deployment/clouddeploy/gke-workers/base/core/exporter.yaml b/deployment/clouddeploy/gke-workers/base/core/exporter.yaml index b76b3a1cffc..f336af9ab96 100644 --- a/deployment/clouddeploy/gke-workers/base/core/exporter.yaml +++ b/deployment/clouddeploy/gke-workers/base/core/exporter.yaml @@ -15,24 +15,29 @@ spec: spec: template: spec: - tolerations: - - key: workloadType - operator: Equal - value: highend - nodeSelector: - cloud.google.com/gke-nodepool: highend containers: - name: exporter image: exporter imagePullPolicy: Always + volumeMounts: + - mountPath: "/scratch" + name: "scratch-volume" resources: requests: - cpu: "12" - memory: "100Gi" + cpu: "4" + memory: "4G" + ephemeral-storage: "15G" limits: - cpu: "20" - memory: "150Gi" + cpu: "7" + memory: "8G" + ephemeral-storage: "30G" env: - name: TRACE_SAMPLE_RATE value: "0.0" + - name: SCRATCH_DIR + value: "/scratch" + volumes: + - name: "scratch-volume" + emptyDir: + sizeLimit: "30G" restartPolicy: Never From 04a2d44eba5c416dfa1f53eb30f4a74829b778ac Mon Sep 17 00:00:00 2001 From: Rex P Date: Wed, 26 Aug 2026 14:44:03 +1000 Subject: [PATCH 04/13] refactor(exporter): simplify scratchDir handling, remove localPath from vulnMeta, and write CSV/Vanir in-memory --- go/cmd/exporter/exporter.go | 31 ++++++-------- go/cmd/exporter/worker.go | 84 ++++++++++++------------------------- 2 files changed, 40 insertions(+), 75 deletions(-) diff --git a/go/cmd/exporter/exporter.go b/go/cmd/exporter/exporter.go index e617a75e6f7..40f3b321838 100644 --- a/go/cmd/exporter/exporter.go +++ b/go/cmd/exporter/exporter.go @@ -1,3 +1,5 @@ +// Package main runs the exporter, exporting the whole OSV database to the GCS bucket. +// See the README.md for more details. package main import ( @@ -52,11 +54,11 @@ func main() { if err := os.MkdirAll(scratchDir, 0755); err != nil { logger.FatalContext(ctx, "failed to create scratch directory", slog.String("dir", scratchDir), slog.Any("err", err)) } - stagingDir, err := os.MkdirTemp(scratchDir, "osv-exporter-staging-*") + scratchDir, err := os.MkdirTemp(scratchDir, "osv-exporter-*") if err != nil { - logger.FatalContext(ctx, "failed to create staging directory in scratch dir", slog.String("dir", scratchDir), slog.Any("err", err)) + logger.FatalContext(ctx, "failed to create temp directory in scratch dir", slog.String("dir", scratchDir), slog.Any("err", err)) } - defer os.RemoveAll(stagingDir) + defer os.RemoveAll(scratchDir) logger.InfoContext(ctx, "exporter starting", slog.String("bucket", *outBucketName), @@ -64,8 +66,7 @@ func main() { slog.Bool("upload-to-gcs", *uploadToGCS), slog.Int("workers", *numWorkers), slog.String("breakdown-prefixes", *breakdownPrefixesStr), - slog.String("scratch-dir", scratchDir), - slog.String("staging-dir", stagingDir)) + slog.String("scratch-dir", scratchDir)) if *vulnBucketName == "" { logger.FatalContext(ctx, "OSV_VULNERABILITIES_BUCKET must be set") @@ -119,7 +120,7 @@ func main() { } var routerWg sync.WaitGroup routerWg.Add(1) - go ecosystemRouter(ctx, downloaderToRouterCh, routerToWriteCh, stagingDir, &routerWg) + go ecosystemRouter(ctx, downloaderToRouterCh, routerToWriteCh, scratchDir, &routerWg) MainLoop: for objName, err := range vulnClient.ObjectsFast(ctx, gcsProtoPrefix, breakdownPrefixes) { @@ -155,12 +156,7 @@ func ecosystemRouter(ctx context.Context, inCh <-chan *osvschema.Vulnerability, workers := make(map[string]*ecosystemWorker) var workersWg sync.WaitGroup vulnCounter := 0 - var vanirVulns []vulnMeta - - vulnsDir := filepath.Join(scratchDir, "vulns") - if err := os.MkdirAll(vulnsDir, 0755); err != nil { - logger.FatalContext(ctx, "failed to create scratch vulns directory", slog.Any("err", err)) - } + var vanirVulns []vulnData allEcosystemWorker := newAllEcosystemWorker(ctx, scratchDir, outCh, &workersWg) @@ -186,23 +182,22 @@ RouterLoop: } // Cache to local scratch disk for later ZIP generation. - localPath := filepath.Join(vulnsDir, vuln.GetId()+".json") + localPath := filepath.Join(scratchDir, vuln.GetId()+".json") if err := os.WriteFile(localPath, b, 0600); err != nil { logger.ErrorContext(ctx, "failed to write cached vulnerability to disk", slog.String("id", vuln.GetId()), slog.Any("err", err)) continue } meta := vulnMeta{ - id: vuln.GetId(), - modified: vuln.GetModified().AsTime(), - localPath: localPath, + id: vuln.GetId(), + modified: vuln.GetModified().AsTime(), } // Check for Vanir signatures for _, aff := range vuln.GetAffected() { spec := aff.GetDatabaseSpecific() if _, ok := spec.GetFields()["vanir_signatures"]; ok { - vanirVulns = append(vanirVulns, meta) + vanirVulns = append(vanirVulns, vulnData{id: vuln.GetId(), data: b}) break } } @@ -256,7 +251,7 @@ RouterLoop: workersWg.Wait() if len(vanirVulns) > 0 && ctx.Err() == nil { - writeVanir(ctx, vanirVulns, outCh, scratchDir) + writeVanir(ctx, vanirVulns, outCh) } if ctx.Err() == nil { diff --git a/go/cmd/exporter/worker.go b/go/cmd/exporter/worker.go index 4d0e2400a02..4c29ab9dbb8 100644 --- a/go/cmd/exporter/worker.go +++ b/go/cmd/exporter/worker.go @@ -31,11 +31,16 @@ const ( ecosystemsFilename = "ecosystems.txt" ) -// vulnMeta holds the ID, modified time, and local staging file path for a vulnerability. +// vulnMeta holds the ID and modified time for a vulnerability. type vulnMeta struct { - id string - modified time.Time - localPath string + id string + modified time.Time +} + +// vulnData holds the ID and marshalled JSON data for a vulnerability. +type vulnData struct { + id string + data []byte } // csvEntry holds the unix timestamp in nanoseconds and the relative entry path. @@ -87,7 +92,7 @@ func (w *ecosystemWorker) run(ctx context.Context, outCh chan<- writeMsg, wg *sy } logger.InfoContext(ctx, "All vulnerabilities processed", slog.String("ecosystem", w.ecosystem)) - writeModifiedIDCSV(ctx, filepath.Join(w.ecosystem, modifiedCSVFilename), csvData, outCh, w.scratchDir) + writeModifiedIDCSV(ctx, filepath.Join(w.ecosystem, modifiedCSVFilename), csvData, outCh) writeZIP(ctx, filepath.Join(w.ecosystem, allZipFilename), allVulns, outCh, w.scratchDir) logger.InfoContext(ctx, "ecosystem worker finished processing", slog.String("ecosystem", w.ecosystem)) } @@ -150,7 +155,7 @@ func (w *allEcosystemWorker) run(ctx context.Context, outCh chan<- writeMsg, wg return } - writeModifiedIDCSV(ctx, modifiedCSVFilename, csvData, outCh, w.scratchDir) + writeModifiedIDCSV(ctx, modifiedCSVFilename, csvData, outCh) writeZIP(ctx, allZipFilename, allVulns, outCh, w.scratchDir) ecos := slices.Collect(maps.Keys(ecosystems)) slices.Sort(ecos) @@ -195,8 +200,8 @@ func writeStream(ctx context.Context, path string, filePath string, mimeType str } } -// writeModifiedIDCSV constructs and writes a modified_id.csv file by streaming to a temporary file. -func writeModifiedIDCSV(ctx context.Context, path string, csvData []csvEntry, outCh chan<- writeMsg, scratchDir string) { +// writeModifiedIDCSV constructs and writes a modified_id.csv file. +func writeModifiedIDCSV(ctx context.Context, path string, csvData []csvEntry, outCh chan<- writeMsg) { logger.InfoContext(ctx, "constructing csv file", slog.String("path", path)) slices.SortFunc(csvData, func(a, b csvEntry) int { return cmp.Or( @@ -205,14 +210,8 @@ func writeModifiedIDCSV(ctx context.Context, path string, csvData []csvEntry, ou ) }) - tmpCsv, err := os.CreateTemp(scratchDir, "csv-*.tmp") - if err != nil { - logger.ErrorContext(ctx, "failed to create temp csv file", slog.String("path", path), slog.Any("err", err)) - return - } - defer tmpCsv.Close() - - wr := csv.NewWriter(tmpCsv) + var buf bytes.Buffer + wr := csv.NewWriter(&buf) for _, entry := range csvData { t := time.Unix(0, entry.modified).UTC().Format(time.RFC3339Nano) if err := wr.Write([]string{t, entry.path}); err != nil { @@ -227,7 +226,7 @@ func writeModifiedIDCSV(ctx context.Context, path string, csvData []csvEntry, ou } logger.InfoContext(ctx, "writing csv file", slog.String("path", path)) - writeStream(ctx, path, tmpCsv.Name(), "text/csv", outCh) + write(ctx, path, buf.Bytes(), "text/csv", outCh) } // writeZIP constructs and writes a zip file by streaming from local files to a temporary zip file. @@ -255,9 +254,10 @@ func writeZIP(ctx context.Context, path string, allVulns []vulnMeta, outCh chan< logger.ErrorContext(ctx, "failed to create vuln json in zip file", slog.String("id", vuln.id), slog.Any("err", err)) continue } - f, err := os.Open(vuln.localPath) + localPath := filepath.Join(scratchDir, vuln.id+".json") + f, err := os.Open(localPath) if err != nil { - logger.ErrorContext(ctx, "failed to open local vuln json file", slog.String("path", vuln.localPath), slog.Any("err", err)) + logger.ErrorContext(ctx, "failed to open local vuln json file", slog.String("path", localPath), slog.Any("err", err)) continue } if _, err := io.Copy(w, f); err != nil { @@ -275,46 +275,16 @@ func writeZIP(ctx context.Context, path string, allVulns []vulnMeta, outCh chan< } // writeVanir constructs and writes the osv_git.json file containing vulnerabilities with Vanir signatures. -func writeVanir(ctx context.Context, vanirVulns []vulnMeta, outCh chan<- writeMsg, scratchDir string) { - slices.SortFunc(vanirVulns, func(a, b vulnMeta) int { return cmp.Compare(a.id, b.id) }) - - tmpVanir, err := os.CreateTemp(scratchDir, "vanir-*.json") - if err != nil { - logger.ErrorContext(ctx, "failed to create temp vanir file", slog.Any("err", err)) - return - } - defer tmpVanir.Close() - - if _, err := tmpVanir.WriteString("[\n"); err != nil { - logger.ErrorContext(ctx, "failed to write vanir header", slog.Any("err", err)) - return - } +func writeVanir(ctx context.Context, vanirVulns []vulnData, outCh chan<- writeMsg) { + slices.SortFunc(vanirVulns, func(a, b vulnData) int { return cmp.Compare(a.id, b.id) }) + vulns := make([]json.RawMessage, len(vanirVulns)) for i, v := range vanirVulns { - data, err := os.ReadFile(v.localPath) - if err != nil { - logger.ErrorContext(ctx, "failed to read local vanir file", slog.String("path", v.localPath), slog.Any("err", err)) - continue - } - if _, err := tmpVanir.Write(data); err != nil { - logger.ErrorContext(ctx, "failed to write vanir entry", slog.Any("err", err)) - return - } - if i < len(vanirVulns)-1 { - if _, err := tmpVanir.WriteString(",\n"); err != nil { - logger.ErrorContext(ctx, "failed to write vanir separator", slog.Any("err", err)) - return - } - } else { - if _, err := tmpVanir.WriteString("\n"); err != nil { - logger.ErrorContext(ctx, "failed to write vanir newline", slog.Any("err", err)) - return - } - } + vulns[i] = v.data } - if _, err := tmpVanir.WriteString("]\n"); err != nil { - logger.ErrorContext(ctx, "failed to write vanir footer", slog.Any("err", err)) + finalJSON, err := json.Marshal(vulns) + if err != nil { + logger.ErrorContext(ctx, "failed to marshal vanir JSON file", slog.Any("err", err)) return } - - writeStream(ctx, filepath.Join(gitEcosystem, vanirVulnsFilename), tmpVanir.Name(), "application/json", outCh) + write(ctx, filepath.Join(gitEcosystem, vanirVulnsFilename), finalJSON, "application/json", outCh) } From bd115f26c8df9c4f8c6689d9c452389a9183104d Mon Sep 17 00:00:00 2001 From: Rex P Date: Wed, 26 Aug 2026 15:10:37 +1000 Subject: [PATCH 05/13] refactor(exporter): parallelize downloading and processing into downloadThenProcessor, stream cleanups, and fast-fail on disk error --- go/cmd/exporter/downloader.go | 77 +++++++++++++++++++++-- go/cmd/exporter/exporter.go | 104 +++++++++---------------------- go/cmd/exporter/exporter_test.go | 63 ++++++++++++------- go/cmd/exporter/worker.go | 7 +++ go/cmd/exporter/writer.go | 6 ++ 5 files changed, 153 insertions(+), 104 deletions(-) diff --git a/go/cmd/exporter/downloader.go b/go/cmd/exporter/downloader.go index 4858c8480b4..ca77bb223f2 100644 --- a/go/cmd/exporter/downloader.go +++ b/go/cmd/exporter/downloader.go @@ -3,6 +3,9 @@ package main import ( "context" "log/slog" + "os" + "path/filepath" + "strings" "sync" "github.com/google/osv.dev/go/internal/osvutil" @@ -12,10 +15,11 @@ import ( "google.golang.org/protobuf/proto" ) -// downloader is a worker that receives GCS object handles from inCh, downloads -// the raw protobuf data, unmarshals it into a Vulnerability, and sends the -// result to outCh. -func downloader(ctx context.Context, client clients.CloudStorage, inCh <-chan string, outCh chan<- *osvschema.Vulnerability, wg *sync.WaitGroup) { +// downloadThenProcessor is a worker that receives GCS object handles from inCh, downloads +// the raw protobuf data, unmarshals it into a Vulnerability, marshals it to compact +// JSON, saves it to scratch disk, queues individual JSON uploads, and sends the +// metadata to routerCh. +func downloadThenProcessor(ctx context.Context, cancel context.CancelFunc, client clients.CloudStorage, scratchDir string, inCh <-chan string, routerCh chan<- processedVuln, writeCh chan<- writeMsg, wg *sync.WaitGroup) { defer wg.Done() for path := range inCh { // Process object. @@ -38,9 +42,70 @@ func downloader(ctx context.Context, client clients.CloudStorage, inCh <-chan st continue } - // Wait to send the result, or be cancelled. + // Marshal JSON ONCE for this vulnerability. + b, err := marshalToJSON(vuln) + if err != nil { + logger.ErrorContext(ctx, "failed to marshal vulnerability to json", slog.String("id", vuln.GetId()), slog.Any("err", err)) + continue + } + + // Cache to local scratch disk for later ZIP generation. + localPath := filepath.Join(scratchDir, vuln.GetId()+".json") + if err := os.WriteFile(localPath, b, 0600); err != nil { + logger.ErrorContext(ctx, "failed to write cached vulnerability to disk", slog.String("id", vuln.GetId()), slog.Any("err", err)) + // Cancel the exporter context if writing to the scratch disk fails (e.g. disk full) + // to fail fast rather than producing incomplete archives later. + cancel() + + return + } + + var vanirData []byte + // Check for Vanir signatures + for _, aff := range vuln.GetAffected() { + spec := aff.GetDatabaseSpecific() + if _, ok := spec.GetFields()["vanir_signatures"]; ok { + vanirData = b + break + } + } + + ecosystems := make(map[string]struct{}) + for _, aff := range vuln.GetAffected() { + eco := aff.GetPackage().GetEcosystem() + eco, _, _ = strings.Cut(eco, ":") + if eco != "" { + ecosystems[eco] = struct{}{} + } + for _, ref := range aff.GetRanges() { + if ref.GetType() == osvschema.Range_GIT { + ecosystems["GIT"] = struct{}{} + } + } + } + if len(ecosystems) == 0 { + ecosystems["[EMPTY]"] = struct{}{} + } + ecoNames := make([]string, 0, len(ecosystems)) + for eco := range ecosystems { + ecoNames = append(ecoNames, eco) + select { + case writeCh <- writeMsg{path: filepath.Join(eco, vuln.GetId()) + ".json", mimeType: "application/json", data: b}: + case <-ctx.Done(): + return + } + } + + // Send processed metadata to router. select { - case outCh <- vuln: + case routerCh <- processedVuln{ + meta: vulnMeta{ + id: vuln.GetId(), + modified: vuln.GetModified().AsTime(), + }, + ecosystems: ecoNames, + vanirData: vanirData, + }: case <-ctx.Done(): return } diff --git a/go/cmd/exporter/exporter.go b/go/cmd/exporter/exporter.go index 40f3b321838..d2a5807ff0c 100644 --- a/go/cmd/exporter/exporter.go +++ b/go/cmd/exporter/exporter.go @@ -9,7 +9,6 @@ import ( "os" "os/signal" "path/filepath" - "strings" "sync" "syscall" @@ -17,7 +16,6 @@ import ( "github.com/google/osv.dev/go/internal/sharding" "github.com/google/osv.dev/go/logger" "github.com/google/osv.dev/go/osv/clients" - "github.com/ossf/osv-schema/bindings/go/osvschema" "go.opentelemetry.io/otel" "google.golang.org/api/option" ) @@ -92,35 +90,36 @@ func main() { } // The exporter uses a pipeline of channels and worker pools. The data flow is as follows: - // 1. The main goroutine lists GCS objects and sends them to `gcsPathToDownloaderCh`. - // 2. A pool of `downloader` workers receive GCS objects, downloads and unmarshals them into - // OSV vulnerabilities, and send them to `downloaderToRouterCh`. - // 3. The `ecosystemRouter` receives vulnerabilities and dispatches them. It creates a new + // 1. The main goroutine lists GCS objects and sends them to `gcsPathToProcessorCh`. + // 2. A pool of `downloadThenProcessor` workers receive GCS objects, downloads, unmarshals, + // marshals to JSON, saves to scratch disk, queues individual JSON writes to `writeCh`, + // and sends metadata to `processorToRouterCh`. + // 3. The `ecosystemRouter` receives metadata and dispatches it. It creates a new // `ecosystemWorker` for each new ecosystem, and sends all vulnerabilities to a single // `allEcosystemWorker`. - // 4. The `ecosystemWorker`s and the `allEcosystemWorker` process the vulnerabilities and - // generate the final files, sending the data to be written to `routerToWriteCh`. + // 4. The `ecosystemWorker`s and the `allEcosystemWorker` aggregate metadata and generate the + // final zip, csv, and ecosystems.txt files, sending the write requests to `writeCh`. // 5. A pool of `writer` workers receive the file data and write it to the output. - gcsPathToDownloaderCh := make(chan string, 100) - downloaderToRouterCh := make(chan *osvschema.Vulnerability, 100) - routerToWriteCh := make(chan writeMsg, 100) + gcsPathToProcessorCh := make(chan string, 100) + processorToRouterCh := make(chan processedVuln, 100) + writeCh := make(chan writeMsg, 100) breakdownPrefixes := sharding.ExpandBreakdownPrefixes(*breakdownPrefixesStr) - var downloaderWg sync.WaitGroup + var processorWg sync.WaitGroup for range *numWorkers / 2 { - downloaderWg.Add(1) - go downloader(ctx, vulnClient, gcsPathToDownloaderCh, downloaderToRouterCh, &downloaderWg) + processorWg.Add(1) + go downloadThenProcessor(ctx, cancel, vulnClient, scratchDir, gcsPathToProcessorCh, processorToRouterCh, writeCh, &processorWg) } var writerWg sync.WaitGroup for range *numWorkers / 2 { writerWg.Add(1) - go writer(ctx, cancel, routerToWriteCh, outClient, outPrefix, &writerWg) + go writer(ctx, cancel, writeCh, outClient, outPrefix, &writerWg) } var routerWg sync.WaitGroup routerWg.Add(1) - go ecosystemRouter(ctx, downloaderToRouterCh, routerToWriteCh, scratchDir, &routerWg) + go ecosystemRouter(ctx, processorToRouterCh, writeCh, scratchDir, &routerWg) MainLoop: for objName, err := range vulnClient.ObjectsFast(ctx, gcsProtoPrefix, breakdownPrefixes) { @@ -128,17 +127,17 @@ MainLoop: logger.FatalContext(ctx, "failed to list objects", slog.Any("err", err)) } select { - case gcsPathToDownloaderCh <- objName: + case gcsPathToProcessorCh <- objName: case <-ctx.Done(): break MainLoop } } - close(gcsPathToDownloaderCh) - downloaderWg.Wait() - close(downloaderToRouterCh) + close(gcsPathToProcessorCh) + processorWg.Wait() + close(processorToRouterCh) routerWg.Wait() - close(routerToWriteCh) + close(writeCh) writerWg.Wait() if ctx.Err() != nil { @@ -147,10 +146,10 @@ MainLoop: logger.InfoContext(ctx, "export completed successfully") } -// ecosystemRouter receives vulnerabilities from inCh and fans them out to the +// ecosystemRouter receives processed vulnerabilities from inCh and fans them out to the // appropriate ecosystemWorker. It creates workers on-demand for each new // ecosystem encountered. It also sends every vulnerability to the allEcosystemWorker. -func ecosystemRouter(ctx context.Context, inCh <-chan *osvschema.Vulnerability, outCh chan<- writeMsg, scratchDir string, wg *sync.WaitGroup) { +func ecosystemRouter(ctx context.Context, inCh <-chan processedVuln, outCh chan<- writeMsg, scratchDir string, wg *sync.WaitGroup) { defer wg.Done() logger.InfoContext(ctx, "ecosystem router starting") workers := make(map[string]*ecosystemWorker) @@ -162,7 +161,7 @@ func ecosystemRouter(ctx context.Context, inCh <-chan *osvschema.Vulnerability, RouterLoop: for { - var vuln *osvschema.Vulnerability + var vuln processedVuln var ok bool select { case <-ctx.Done(): @@ -174,71 +173,24 @@ RouterLoop: } vulnCounter++ - // Marshal JSON ONCE for this vulnerability. - b, err := marshalToJSON(vuln) - if err != nil { - logger.ErrorContext(ctx, "failed to marshal vulnerability to json", slog.String("id", vuln.GetId()), slog.Any("err", err)) - continue - } - - // Cache to local scratch disk for later ZIP generation. - localPath := filepath.Join(scratchDir, vuln.GetId()+".json") - if err := os.WriteFile(localPath, b, 0600); err != nil { - logger.ErrorContext(ctx, "failed to write cached vulnerability to disk", slog.String("id", vuln.GetId()), slog.Any("err", err)) - continue - } - - meta := vulnMeta{ - id: vuln.GetId(), - modified: vuln.GetModified().AsTime(), - } - - // Check for Vanir signatures - for _, aff := range vuln.GetAffected() { - spec := aff.GetDatabaseSpecific() - if _, ok := spec.GetFields()["vanir_signatures"]; ok { - vanirVulns = append(vanirVulns, vulnData{id: vuln.GetId(), data: b}) - break - } + if len(vuln.vanirData) > 0 { + vanirVulns = append(vanirVulns, vulnData{id: vuln.meta.id, data: vuln.vanirData}) } - ecosystems := make(map[string]struct{}) - for _, aff := range vuln.GetAffected() { - eco := aff.GetPackage().GetEcosystem() - eco, _, _ = strings.Cut(eco, ":") - if eco != "" { - ecosystems[eco] = struct{}{} - } - for _, ref := range aff.GetRanges() { - if ref.GetType() == osvschema.Range_GIT { - ecosystems["GIT"] = struct{}{} - } - } - } - if len(ecosystems) == 0 { - ecosystems["[EMPTY]"] = struct{}{} - } - ecoNames := make([]string, 0, len(ecosystems)) - for eco := range ecosystems { - ecoNames = append(ecoNames, eco) + for _, eco := range vuln.ecosystems { worker, ok := workers[eco] if !ok { worker = newEcosystemWorker(ctx, eco, scratchDir, outCh, &workersWg) workers[eco] = worker } select { - case worker.inCh <- meta: - case <-ctx.Done(): - break RouterLoop - } - select { - case outCh <- writeMsg{path: filepath.Join(eco, vuln.GetId()) + ".json", mimeType: "application/json", data: b}: + case worker.inCh <- vuln.meta: case <-ctx.Done(): break RouterLoop } } select { - case allEcosystemWorker.inCh <- vulnAndEcos{meta: meta, ecosystems: ecoNames}: + case allEcosystemWorker.inCh <- vulnAndEcos{meta: vuln.meta, ecosystems: vuln.ecosystems}: case <-ctx.Done(): break RouterLoop } diff --git a/go/cmd/exporter/exporter_test.go b/go/cmd/exporter/exporter_test.go index db6d5a4a350..813933ad9b8 100644 --- a/go/cmd/exporter/exporter_test.go +++ b/go/cmd/exporter/exporter_test.go @@ -14,6 +14,7 @@ import ( "github.com/google/osv.dev/go/testutils" "github.com/ossf/osv-schema/bindings/go/osvschema" + "google.golang.org/protobuf/proto" "google.golang.org/protobuf/types/known/structpb" "google.golang.org/protobuf/types/known/timestamppb" ) @@ -74,18 +75,8 @@ func TestExporterPipeline_EndToEnd(t *testing.T) { defer cancel() scratchDir := t.TempDir() - storage := testutils.NewMockStorage() - - inCh := make(chan *osvschema.Vulnerability, 10) - routerToWriteCh := make(chan writeMsg, 100) - - var writerWg sync.WaitGroup - writerWg.Add(1) - go writer(ctx, cancel, routerToWriteCh, storage, "export", &writerWg) - - var routerWg sync.WaitGroup - routerWg.Add(1) - go ecosystemRouter(ctx, inCh, routerToWriteCh, scratchDir, &routerWg) + vulnStorage := testutils.NewMockStorage() + outStorage := testutils.NewMockStorage() time1 := time.Date(2023, 1, 1, 12, 0, 0, 0, time.UTC) time2 := time.Date(2023, 2, 1, 12, 0, 0, 0, time.UTC) @@ -103,6 +94,11 @@ func TestExporterPipeline_EndToEnd(t *testing.T) { }, }, } + pb1, err := proto.Marshal(vuln1) + if err != nil { + t.Fatalf("failed to marshal proto: %v", err) + } + _ = vulnStorage.WriteObject(ctx, "all/pb/vuln1", pb1, nil) // Create test vulnerability 2: npm and GIT with Vanir signatures vanirField, _ := structpb.NewValue("test-signature") @@ -128,17 +124,40 @@ func TestExporterPipeline_EndToEnd(t *testing.T) { }, }, } + pb2, err := proto.Marshal(vuln2) + if err != nil { + t.Fatalf("failed to marshal proto: %v", err) + } + _ = vulnStorage.WriteObject(ctx, "all/pb/vuln2", pb2, nil) + + gcsPathToProcessorCh := make(chan string, 10) + processorToRouterCh := make(chan processedVuln, 10) + writeCh := make(chan writeMsg, 100) + + var processorWg sync.WaitGroup + processorWg.Add(1) + go downloadThenProcessor(ctx, cancel, vulnStorage, scratchDir, gcsPathToProcessorCh, processorToRouterCh, writeCh, &processorWg) + + var writerWg sync.WaitGroup + writerWg.Add(1) + go writer(ctx, cancel, writeCh, outStorage, "export", &writerWg) + + var routerWg sync.WaitGroup + routerWg.Add(1) + go ecosystemRouter(ctx, processorToRouterCh, writeCh, scratchDir, &routerWg) - inCh <- vuln1 - inCh <- vuln2 - close(inCh) + gcsPathToProcessorCh <- "all/pb/vuln1" + gcsPathToProcessorCh <- "all/pb/vuln2" + close(gcsPathToProcessorCh) + processorWg.Wait() + close(processorToRouterCh) routerWg.Wait() - close(routerToWriteCh) + close(writeCh) writerWg.Wait() // 1. Verify individual JSON outputs - pypiJSON, err := storage.ReadObject(ctx, "export/PyPI/GHSA-pypi-1.json") + pypiJSON, err := outStorage.ReadObject(ctx, "export/PyPI/GHSA-pypi-1.json") if err != nil { t.Fatalf("expected PyPI/GHSA-pypi-1.json in storage: %v", err) } @@ -146,7 +165,7 @@ func TestExporterPipeline_EndToEnd(t *testing.T) { t.Errorf("PyPI JSON content mismatch: %s", string(pypiJSON)) } - npmJSON, err := storage.ReadObject(ctx, "export/npm/GHSA-npm-git-2.json") + npmJSON, err := outStorage.ReadObject(ctx, "export/npm/GHSA-npm-git-2.json") if err != nil { t.Fatalf("expected npm/GHSA-npm-git-2.json in storage: %v", err) } @@ -155,7 +174,7 @@ func TestExporterPipeline_EndToEnd(t *testing.T) { } // 2. Verify all.zip contains all JSON files - allZipBytes, err := storage.ReadObject(ctx, "export/all.zip") + allZipBytes, err := outStorage.ReadObject(ctx, "export/all.zip") if err != nil { t.Fatalf("expected all.zip: %v", err) } @@ -172,7 +191,7 @@ func TestExporterPipeline_EndToEnd(t *testing.T) { } // 3. Verify modified_id.csv ordering (descending by modified time) - csvBytes, err := storage.ReadObject(ctx, "export/modified_id.csv") + csvBytes, err := outStorage.ReadObject(ctx, "export/modified_id.csv") if err != nil { t.Fatalf("expected modified_id.csv: %v", err) } @@ -190,7 +209,7 @@ func TestExporterPipeline_EndToEnd(t *testing.T) { } // 4. Verify ecosystems.txt - ecoTxtBytes, err := storage.ReadObject(ctx, "export/ecosystems.txt") + ecoTxtBytes, err := outStorage.ReadObject(ctx, "export/ecosystems.txt") if err != nil { t.Fatalf("expected ecosystems.txt: %v", err) } @@ -200,7 +219,7 @@ func TestExporterPipeline_EndToEnd(t *testing.T) { } // 5. Verify Vanir signatures file (GIT/osv_git.json) - vanirBytes, err := storage.ReadObject(ctx, "export/GIT/osv_git.json") + vanirBytes, err := outStorage.ReadObject(ctx, "export/GIT/osv_git.json") if err != nil { t.Fatalf("expected GIT/osv_git.json: %v", err) } diff --git a/go/cmd/exporter/worker.go b/go/cmd/exporter/worker.go index 4c29ab9dbb8..1ca24e2c223 100644 --- a/go/cmd/exporter/worker.go +++ b/go/cmd/exporter/worker.go @@ -49,6 +49,13 @@ type csvEntry struct { path string } +// processedVuln holds the metadata, ecosystems, and optional Vanir signatures for a processed vulnerability. +type processedVuln struct { + meta vulnMeta + ecosystems []string + vanirData []byte +} + // ecosystemWorker processes vulnerabilities for a single ecosystem. type ecosystemWorker struct { ecosystem string diff --git a/go/cmd/exporter/writer.go b/go/cmd/exporter/writer.go index a10d550e03d..79723951cd4 100644 --- a/go/cmd/exporter/writer.go +++ b/go/cmd/exporter/writer.go @@ -34,11 +34,13 @@ func writer(ctx context.Context, cancel context.CancelFunc, inCh <-chan writeMsg if msg.filePath != "" { if client != nil { if gcsFileUnchanged(ctx, client, path, msg.filePath) { + _ = os.Remove(msg.filePath) continue } f, err := os.Open(msg.filePath) if err != nil { logger.Error("failed to open local file for upload", slog.String("path", path), slog.String("file", msg.filePath), slog.Any("err", err)) + _ = os.Remove(msg.filePath) cancel() break @@ -47,6 +49,7 @@ func writer(ctx context.Context, cancel context.CancelFunc, inCh <-chan writeMsg ContentType: msg.mimeType, }) f.Close() + _ = os.Remove(msg.filePath) if err != nil { logger.Error("failed to stream write file", slog.String("path", path), slog.Any("err", err)) cancel() @@ -58,16 +61,19 @@ func writer(ctx context.Context, cancel context.CancelFunc, inCh <-chan writeMsg dir := filepath.Dir(path) if err := os.MkdirAll(dir, 0755); err != nil { logger.Error("failed to create directories", slog.String("dir", dir), slog.Any("err", err)) + _ = os.Remove(msg.filePath) cancel() break } if err := copyFile(msg.filePath, path); err != nil { logger.Error("failed to copy file locally", slog.String("src", msg.filePath), slog.String("dst", path), slog.Any("err", err)) + _ = os.Remove(msg.filePath) cancel() break } + _ = os.Remove(msg.filePath) } } else { if client != nil { From 1f37b688c23cae333be2c6e02a21834bae5ad498 Mon Sep 17 00:00:00 2001 From: Rex P Date: Wed, 26 Aug 2026 15:13:23 +1000 Subject: [PATCH 06/13] refactor(exporter): stream Vanir signature json generation directly from scratch disk --- go/cmd/exporter/downloader.go | 6 ++-- go/cmd/exporter/exporter.go | 10 +++--- go/cmd/exporter/worker.go | 66 +++++++++++++++++++++++++---------- 3 files changed, 56 insertions(+), 26 deletions(-) diff --git a/go/cmd/exporter/downloader.go b/go/cmd/exporter/downloader.go index ca77bb223f2..0aba4c7a10f 100644 --- a/go/cmd/exporter/downloader.go +++ b/go/cmd/exporter/downloader.go @@ -60,12 +60,12 @@ func downloadThenProcessor(ctx context.Context, cancel context.CancelFunc, clien return } - var vanirData []byte + hasVanir := false // Check for Vanir signatures for _, aff := range vuln.GetAffected() { spec := aff.GetDatabaseSpecific() if _, ok := spec.GetFields()["vanir_signatures"]; ok { - vanirData = b + hasVanir = true break } } @@ -104,7 +104,7 @@ func downloadThenProcessor(ctx context.Context, cancel context.CancelFunc, clien modified: vuln.GetModified().AsTime(), }, ecosystems: ecoNames, - vanirData: vanirData, + hasVanir: hasVanir, }: case <-ctx.Done(): return diff --git a/go/cmd/exporter/exporter.go b/go/cmd/exporter/exporter.go index d2a5807ff0c..acf2fbeb29b 100644 --- a/go/cmd/exporter/exporter.go +++ b/go/cmd/exporter/exporter.go @@ -155,7 +155,7 @@ func ecosystemRouter(ctx context.Context, inCh <-chan processedVuln, outCh chan< workers := make(map[string]*ecosystemWorker) var workersWg sync.WaitGroup vulnCounter := 0 - var vanirVulns []vulnData + var vanirVulnIDs []string allEcosystemWorker := newAllEcosystemWorker(ctx, scratchDir, outCh, &workersWg) @@ -173,8 +173,8 @@ RouterLoop: } vulnCounter++ - if len(vuln.vanirData) > 0 { - vanirVulns = append(vanirVulns, vulnData{id: vuln.meta.id, data: vuln.vanirData}) + if vuln.hasVanir { + vanirVulnIDs = append(vanirVulnIDs, vuln.meta.id) } for _, eco := range vuln.ecosystems { @@ -202,8 +202,8 @@ RouterLoop: allEcosystemWorker.Finish() workersWg.Wait() - if len(vanirVulns) > 0 && ctx.Err() == nil { - writeVanir(ctx, vanirVulns, outCh) + if len(vanirVulnIDs) > 0 && ctx.Err() == nil { + writeVanir(ctx, vanirVulnIDs, outCh, scratchDir) } if ctx.Err() == nil { diff --git a/go/cmd/exporter/worker.go b/go/cmd/exporter/worker.go index 1ca24e2c223..ff5ab622053 100644 --- a/go/cmd/exporter/worker.go +++ b/go/cmd/exporter/worker.go @@ -37,23 +37,17 @@ type vulnMeta struct { modified time.Time } -// vulnData holds the ID and marshalled JSON data for a vulnerability. -type vulnData struct { - id string - data []byte -} - // csvEntry holds the unix timestamp in nanoseconds and the relative entry path. type csvEntry struct { modified int64 path string } -// processedVuln holds the metadata, ecosystems, and optional Vanir signatures for a processed vulnerability. +// processedVuln holds the metadata, ecosystems, and Vanir flag for a processed vulnerability. type processedVuln struct { meta vulnMeta ecosystems []string - vanirData []byte + hasVanir bool } // ecosystemWorker processes vulnerabilities for a single ecosystem. @@ -281,17 +275,53 @@ func writeZIP(ctx context.Context, path string, allVulns []vulnMeta, outCh chan< writeStream(ctx, path, tmpZip.Name(), "application/zip", outCh) } -// writeVanir constructs and writes the osv_git.json file containing vulnerabilities with Vanir signatures. -func writeVanir(ctx context.Context, vanirVulns []vulnData, outCh chan<- writeMsg) { - slices.SortFunc(vanirVulns, func(a, b vulnData) int { return cmp.Compare(a.id, b.id) }) - vulns := make([]json.RawMessage, len(vanirVulns)) - for i, v := range vanirVulns { - vulns[i] = v.data - } - finalJSON, err := json.Marshal(vulns) +// writeVanir constructs and writes the osv_git.json file containing vulnerabilities with Vanir signatures +// by streaming the cached JSON files from disk into a temporary JSON array file. +func writeVanir(ctx context.Context, vanirVulnIDs []string, outCh chan<- writeMsg, scratchDir string) { + logger.InfoContext(ctx, "constructing vanir file", slog.Int("count", len(vanirVulnIDs))) + slices.Sort(vanirVulnIDs) + + tmpVanir, err := os.CreateTemp(scratchDir, "vanir-*.json") if err != nil { - logger.ErrorContext(ctx, "failed to marshal vanir JSON file", slog.Any("err", err)) + logger.ErrorContext(ctx, "failed to create temp vanir file", slog.Any("err", err)) + return + } + defer tmpVanir.Close() + + if _, err := tmpVanir.WriteString("["); err != nil { + logger.ErrorContext(ctx, "failed to write vanir header", slog.Any("err", err)) + return + } + + first := true + for _, id := range vanirVulnIDs { + localPath := filepath.Join(scratchDir, id+".json") + f, err := os.Open(localPath) + if err != nil { + logger.ErrorContext(ctx, "failed to open local vuln file for vanir", slog.String("id", id), slog.Any("err", err)) + continue + } + if !first { + if _, err := tmpVanir.WriteString(","); err != nil { + f.Close() + logger.ErrorContext(ctx, "failed to write vanir separator", slog.Any("err", err)) + return + } + } + if _, err := io.Copy(tmpVanir, f); err != nil { + f.Close() + logger.ErrorContext(ctx, "failed to copy vuln to vanir file", slog.String("id", id), slog.Any("err", err)) + return + } + f.Close() + first = false + } + + if _, err := tmpVanir.WriteString("]"); err != nil { + logger.ErrorContext(ctx, "failed to write vanir footer", slog.Any("err", err)) return } - write(ctx, filepath.Join(gitEcosystem, vanirVulnsFilename), finalJSON, "application/json", outCh) + + logger.InfoContext(ctx, "writing vanir file", slog.String("path", filepath.Join(gitEcosystem, vanirVulnsFilename))) + writeStream(ctx, filepath.Join(gitEcosystem, vanirVulnsFilename), tmpVanir.Name(), "application/json", outCh) } From 09c45a59fb2e7eff13060bc5e8094ed929b762a2 Mon Sep 17 00:00:00 2001 From: Rex P Date: Wed, 26 Aug 2026 15:20:18 +1000 Subject: [PATCH 07/13] refactor(exporter): build Vanir signature json in memory from disk files --- go/cmd/exporter/worker.go | 42 +++++++++------------------------------ 1 file changed, 9 insertions(+), 33 deletions(-) diff --git a/go/cmd/exporter/worker.go b/go/cmd/exporter/worker.go index ff5ab622053..f699b909a94 100644 --- a/go/cmd/exporter/worker.go +++ b/go/cmd/exporter/worker.go @@ -276,52 +276,28 @@ func writeZIP(ctx context.Context, path string, allVulns []vulnMeta, outCh chan< } // writeVanir constructs and writes the osv_git.json file containing vulnerabilities with Vanir signatures -// by streaming the cached JSON files from disk into a temporary JSON array file. +// by reading the cached JSON files from disk and marshaling the combined JSON array in memory. func writeVanir(ctx context.Context, vanirVulnIDs []string, outCh chan<- writeMsg, scratchDir string) { logger.InfoContext(ctx, "constructing vanir file", slog.Int("count", len(vanirVulnIDs))) slices.Sort(vanirVulnIDs) - tmpVanir, err := os.CreateTemp(scratchDir, "vanir-*.json") - if err != nil { - logger.ErrorContext(ctx, "failed to create temp vanir file", slog.Any("err", err)) - return - } - defer tmpVanir.Close() - - if _, err := tmpVanir.WriteString("["); err != nil { - logger.ErrorContext(ctx, "failed to write vanir header", slog.Any("err", err)) - return - } - - first := true + vulns := make([]json.RawMessage, 0, len(vanirVulnIDs)) for _, id := range vanirVulnIDs { localPath := filepath.Join(scratchDir, id+".json") - f, err := os.Open(localPath) + data, err := os.ReadFile(localPath) if err != nil { - logger.ErrorContext(ctx, "failed to open local vuln file for vanir", slog.String("id", id), slog.Any("err", err)) + logger.ErrorContext(ctx, "failed to read local vuln file for vanir", slog.String("id", id), slog.Any("err", err)) continue } - if !first { - if _, err := tmpVanir.WriteString(","); err != nil { - f.Close() - logger.ErrorContext(ctx, "failed to write vanir separator", slog.Any("err", err)) - return - } - } - if _, err := io.Copy(tmpVanir, f); err != nil { - f.Close() - logger.ErrorContext(ctx, "failed to copy vuln to vanir file", slog.String("id", id), slog.Any("err", err)) - return - } - f.Close() - first = false + vulns = append(vulns, data) } - if _, err := tmpVanir.WriteString("]"); err != nil { - logger.ErrorContext(ctx, "failed to write vanir footer", slog.Any("err", err)) + finalJSON, err := json.Marshal(vulns) + if err != nil { + logger.ErrorContext(ctx, "failed to marshal vanir JSON file", slog.Any("err", err)) return } logger.InfoContext(ctx, "writing vanir file", slog.String("path", filepath.Join(gitEcosystem, vanirVulnsFilename))) - writeStream(ctx, filepath.Join(gitEcosystem, vanirVulnsFilename), tmpVanir.Name(), "application/json", outCh) + write(ctx, filepath.Join(gitEcosystem, vanirVulnsFilename), finalJSON, "application/json", outCh) } From 5117f805ea06a7c02a4c028b5bd2e8e36d8b6faa Mon Sep 17 00:00:00 2001 From: Rex P Date: Wed, 26 Aug 2026 15:28:04 +1000 Subject: [PATCH 08/13] refactor(exporter): split writer into writeFromFile and writeFromMemory helper functions --- go/cmd/exporter/writer.go | 132 +++++++++++++++++--------------------- 1 file changed, 59 insertions(+), 73 deletions(-) diff --git a/go/cmd/exporter/writer.go b/go/cmd/exporter/writer.go index 79723951cd4..8dbaf62ab8b 100644 --- a/go/cmd/exporter/writer.go +++ b/go/cmd/exporter/writer.go @@ -31,83 +31,69 @@ func writer(ctx context.Context, cancel context.CancelFunc, inCh <-chan writeMsg defer wg.Done() for msg := range inCh { path := filepath.Join(pathPrefix, msg.path) + var err error if msg.filePath != "" { - if client != nil { - if gcsFileUnchanged(ctx, client, path, msg.filePath) { - _ = os.Remove(msg.filePath) - continue - } - f, err := os.Open(msg.filePath) - if err != nil { - logger.Error("failed to open local file for upload", slog.String("path", path), slog.String("file", msg.filePath), slog.Any("err", err)) - _ = os.Remove(msg.filePath) - cancel() - - break - } - err = client.WriteObjectStream(ctx, path, f, &clients.WriteOptions{ - ContentType: msg.mimeType, - }) - f.Close() - _ = os.Remove(msg.filePath) - if err != nil { - logger.Error("failed to stream write file", slog.String("path", path), slog.Any("err", err)) - cancel() - - break - } - } else { - // Write locally: copy from msg.filePath to path - dir := filepath.Dir(path) - if err := os.MkdirAll(dir, 0755); err != nil { - logger.Error("failed to create directories", slog.String("dir", dir), slog.Any("err", err)) - _ = os.Remove(msg.filePath) - cancel() - - break - } - if err := copyFile(msg.filePath, path); err != nil { - logger.Error("failed to copy file locally", slog.String("src", msg.filePath), slog.String("dst", path), slog.Any("err", err)) - _ = os.Remove(msg.filePath) - cancel() - - break - } - _ = os.Remove(msg.filePath) - } + err = writeFromFile(ctx, client, path, msg.filePath, msg.mimeType) } else { - if client != nil { - // Skip the upload if the object already has the same content. - if gcsContentUnchanged(ctx, client, path, msg.data) { - continue - } - err := client.WriteObject(ctx, path, msg.data, &clients.WriteOptions{ - ContentType: msg.mimeType, - }) - if err != nil { - logger.Error("failed to write file", slog.String("path", path), slog.Any("err", err)) - cancel() - - break - } - } else { - // Write locally. - dir := filepath.Dir(path) - if err := os.MkdirAll(dir, 0755); err != nil { - logger.Error("failed to create directories", slog.String("dir", dir), slog.Any("err", err)) - cancel() - - break - } - if err := os.WriteFile(path, msg.data, 0600); err != nil { - logger.Error("failed to write file", slog.String("path", path), slog.Any("err", err)) - cancel() - - break - } - } + err = writeFromMemory(ctx, client, path, msg.data, msg.mimeType) } + if err != nil { + logger.Error("failed to write output file", slog.String("path", path), slog.Any("err", err)) + cancel() + + break + } + } +} + +// writeFromFile handles writing or uploading a file by streaming from a local file path. +// It removes the source file when finished. +func writeFromFile(ctx context.Context, client clients.CloudStorage, path, filePath, mimeType string) error { + defer os.Remove(filePath) + + if client != nil { + if gcsFileUnchanged(ctx, client, path, filePath) { + return nil + } + f, err := os.Open(filePath) + if err != nil { + return err + } + defer f.Close() + + return client.WriteObjectStream(ctx, path, f, &clients.WriteOptions{ + ContentType: mimeType, + }) + } + + // Write locally: copy from filePath to path + dir := filepath.Dir(path) + if err := os.MkdirAll(dir, 0755); err != nil { + return err } + + return copyFile(filePath, path) +} + +// writeFromMemory handles writing or uploading in-memory byte data to GCS or a local file. +func writeFromMemory(ctx context.Context, client clients.CloudStorage, path string, data []byte, mimeType string) error { + if client != nil { + if gcsContentUnchanged(ctx, client, path, data) { + return nil + } + + return client.WriteObject(ctx, path, data, &clients.WriteOptions{ + ContentType: mimeType, + }) + } + + // Write locally + dir := filepath.Dir(path) + if err := os.MkdirAll(dir, 0755); err != nil { + return err + } + + return os.WriteFile(path, data, 0600) } func copyFile(src, dst string) error { From e091a6bb684ea7775dfb982a358305a9b9df4a76 Mon Sep 17 00:00:00 2001 From: Rex P Date: Wed, 26 Aug 2026 15:43:50 +1000 Subject: [PATCH 09/13] test(exporter): fix slice preallocation lint finding and document go linter in AGENTS.md --- AGENTS.md | 11 ++++++++++- go/cmd/exporter/exporter_test.go | 2 +- 2 files changed, 11 insertions(+), 2 deletions(-) diff --git a/AGENTS.md b/AGENTS.md index eaa8ed5214b..6695b8c682c 100644 --- a/AGENTS.md +++ b/AGENTS.md @@ -105,7 +105,16 @@ Always format and lint your code before proposing changes. The repository provid - **Rule**: When running Python scripts, always use `poetry run`. ### Go Standards -- Linter: `golangci-lint` (run automatically by the lint script per module). +- Linter: `golangci-lint` +- **Running Go Linters**: Run `golangci-lint` using `go run` directly within the module directory (`go/`, `vulnfeeds/`, or `bindings/go/`): + ```bash + cd go && go run github.com/golangci/golangci-lint/v2/cmd/golangci-lint@v2.4.0 run ./... + ``` + *(Note: Run outside the sandbox so `go run` can fetch the linter toolchain if not cached).* +- **Formatting Command**: + ```bash + cd go && gofmt -s -w . + ``` - **Rule**: Go code must follow standard Go formatting guidelines. ### Git Commit Guidelines diff --git a/go/cmd/exporter/exporter_test.go b/go/cmd/exporter/exporter_test.go index 813933ad9b8..5d2f233fcfb 100644 --- a/go/cmd/exporter/exporter_test.go +++ b/go/cmd/exporter/exporter_test.go @@ -182,7 +182,7 @@ func TestExporterPipeline_EndToEnd(t *testing.T) { if err != nil { t.Fatalf("failed to open all.zip: %v", err) } - var zipNames []string + zipNames := make([]string, 0, len(zipReader.File)) for _, f := range zipReader.File { zipNames = append(zipNames, f.Name) } From 6fe952e5f7b4e88081cdce42836b9eb3c72ae4ee Mon Sep 17 00:00:00 2001 From: Rex P Date: Wed, 26 Aug 2026 16:17:58 +1000 Subject: [PATCH 10/13] refactor(exporter): use time.Time in csvEntry struct and CSV sorting/formatting --- go/cmd/exporter/worker.go | 14 +++++++------- 1 file changed, 7 insertions(+), 7 deletions(-) diff --git a/go/cmd/exporter/worker.go b/go/cmd/exporter/worker.go index f699b909a94..e5aab4e3243 100644 --- a/go/cmd/exporter/worker.go +++ b/go/cmd/exporter/worker.go @@ -37,9 +37,9 @@ type vulnMeta struct { modified time.Time } -// csvEntry holds the unix timestamp in nanoseconds and the relative entry path. +// csvEntry holds the modified time and the relative entry path. type csvEntry struct { - modified int64 + modified time.Time path string } @@ -85,7 +85,7 @@ func (w *ecosystemWorker) run(ctx context.Context, outCh chan<- writeMsg, wg *sy csvData := make([]csvEntry, 0, 500) for v := range w.inCh { allVulns = append(allVulns, v) - csvData = append(csvData, csvEntry{modified: v.modified.UnixNano(), path: v.id}) + csvData = append(csvData, csvEntry{modified: v.modified, path: v.id}) } if ctx.Err() != nil { @@ -145,7 +145,7 @@ func (w *allEcosystemWorker) run(ctx context.Context, outCh chan<- writeMsg, wg allVulns = append(allVulns, v.meta) for _, e := range v.ecosystems { ecosystems[e] = struct{}{} - csvData = append(csvData, csvEntry{modified: v.meta.modified.UnixNano(), path: e + "/" + v.meta.id}) + csvData = append(csvData, csvEntry{modified: v.meta.modified, path: e + "/" + v.meta.id}) if len(csvData)%50000 == 0 { logger.InfoContext(ctx, "processed N vulnerabilities", slog.Int("n", len(csvData))) } @@ -206,15 +206,15 @@ func writeModifiedIDCSV(ctx context.Context, path string, csvData []csvEntry, ou logger.InfoContext(ctx, "constructing csv file", slog.String("path", path)) slices.SortFunc(csvData, func(a, b csvEntry) int { return cmp.Or( - -cmp.Compare(a.modified, b.modified), // Modified date, descending - cmp.Compare(a.path, b.path), // path/vuln ID, ascending + -a.modified.Compare(b.modified), // Modified date, descending + cmp.Compare(a.path, b.path), // path/vuln ID, ascending ) }) var buf bytes.Buffer wr := csv.NewWriter(&buf) for _, entry := range csvData { - t := time.Unix(0, entry.modified).UTC().Format(time.RFC3339Nano) + t := entry.modified.UTC().Format(time.RFC3339Nano) if err := wr.Write([]string{t, entry.path}); err != nil { logger.ErrorContext(ctx, "failed writing csv line", slog.String("path", path), slog.Any("err", err)) return From 607ba3d607b5d6b8a53438cc5e69da94210e4771 Mon Sep 17 00:00:00 2001 From: Rex P Date: Wed, 26 Aug 2026 16:36:05 +1000 Subject: [PATCH 11/13] refactor(exporter): address PR review comments - add range break and remove file deletion in writer --- go/cmd/exporter/downloader.go | 1 + go/cmd/exporter/writer.go | 3 --- 2 files changed, 1 insertion(+), 3 deletions(-) diff --git a/go/cmd/exporter/downloader.go b/go/cmd/exporter/downloader.go index 0aba4c7a10f..0b66c3d3e24 100644 --- a/go/cmd/exporter/downloader.go +++ b/go/cmd/exporter/downloader.go @@ -80,6 +80,7 @@ func downloadThenProcessor(ctx context.Context, cancel context.CancelFunc, clien for _, ref := range aff.GetRanges() { if ref.GetType() == osvschema.Range_GIT { ecosystems["GIT"] = struct{}{} + break } } } diff --git a/go/cmd/exporter/writer.go b/go/cmd/exporter/writer.go index 8dbaf62ab8b..d9506b18c66 100644 --- a/go/cmd/exporter/writer.go +++ b/go/cmd/exporter/writer.go @@ -47,10 +47,7 @@ func writer(ctx context.Context, cancel context.CancelFunc, inCh <-chan writeMsg } // writeFromFile handles writing or uploading a file by streaming from a local file path. -// It removes the source file when finished. func writeFromFile(ctx context.Context, client clients.CloudStorage, path, filePath, mimeType string) error { - defer os.Remove(filePath) - if client != nil { if gcsFileUnchanged(ctx, client, path, filePath) { return nil From 9167b59de3ba29366fab21c84a1219893f2ea129 Mon Sep 17 00:00:00 2001 From: Rex P Date: Thu, 27 Aug 2026 10:50:57 +1000 Subject: [PATCH 12/13] perf(deploy): set exporter ephemeral storage request to 35G and limit to 50G --- deployment/clouddeploy/gke-workers/base/core/exporter.yaml | 6 +++--- 1 file changed, 3 insertions(+), 3 deletions(-) diff --git a/deployment/clouddeploy/gke-workers/base/core/exporter.yaml b/deployment/clouddeploy/gke-workers/base/core/exporter.yaml index f336af9ab96..42dbe7c1370 100644 --- a/deployment/clouddeploy/gke-workers/base/core/exporter.yaml +++ b/deployment/clouddeploy/gke-workers/base/core/exporter.yaml @@ -26,11 +26,11 @@ spec: requests: cpu: "4" memory: "4G" - ephemeral-storage: "15G" + ephemeral-storage: "35G" limits: cpu: "7" memory: "8G" - ephemeral-storage: "30G" + ephemeral-storage: "50G" env: - name: TRACE_SAMPLE_RATE value: "0.0" @@ -39,5 +39,5 @@ spec: volumes: - name: "scratch-volume" emptyDir: - sizeLimit: "30G" + sizeLimit: "50G" restartPolicy: Never From 5b236551bc890e3d4fbf3bae9c083572efb220e6 Mon Sep 17 00:00:00 2001 From: Rex P Date: Thu, 27 Aug 2026 11:21:33 +1000 Subject: [PATCH 13/13] feat(exporter): log scratch filesystem usage metrics across export stages --- go/cmd/exporter/exporter.go | 30 ++++++++++++++++++++++++++++++ 1 file changed, 30 insertions(+) diff --git a/go/cmd/exporter/exporter.go b/go/cmd/exporter/exporter.go index acf2fbeb29b..000136c22a8 100644 --- a/go/cmd/exporter/exporter.go +++ b/go/cmd/exporter/exporter.go @@ -121,6 +121,8 @@ func main() { routerWg.Add(1) go ecosystemRouter(ctx, processorToRouterCh, writeCh, scratchDir, &routerWg) + logDiskUsage(ctx, scratchDir, "startup") + MainLoop: for objName, err := range vulnClient.ObjectsFast(ctx, gcsProtoPrefix, breakdownPrefixes) { if err != nil { @@ -135,8 +137,10 @@ MainLoop: close(gcsPathToProcessorCh) processorWg.Wait() + logDiskUsage(ctx, scratchDir, "downloads_complete") close(processorToRouterCh) routerWg.Wait() + logDiskUsage(ctx, scratchDir, "export_complete") close(writeCh) writerWg.Wait() @@ -146,6 +150,32 @@ MainLoop: logger.InfoContext(ctx, "export completed successfully") } +// logDiskUsage logs the scratch filesystem space metrics. +func logDiskUsage(ctx context.Context, dir string, stage string) { + var stat syscall.Statfs_t + if err := syscall.Statfs(dir, &stat); err != nil { + logger.WarnContext(ctx, "failed to get scratch disk usage", slog.String("dir", dir), slog.Any("err", err)) + return + } + if stat.Bsize <= 0 || stat.Blocks == 0 { + return + } + + const gib = 1024 * 1024 * 1024 + blockSize := float64(stat.Bsize) + usedGB := float64(stat.Blocks-stat.Bfree) * blockSize / gib + totalGB := float64(stat.Blocks) * blockSize / gib + freeGB := float64(stat.Bavail) * blockSize / gib + + logger.InfoContext(ctx, "scratch disk usage", + slog.String("stage", stage), + slog.Float64("used_gb", usedGB), + slog.Float64("free_gb", freeGB), + slog.Float64("total_gb", totalGB), + slog.Float64("used_pct", usedGB/totalGB*100), + ) +} + // ecosystemRouter receives processed vulnerabilities from inCh and fans them out to the // appropriate ecosystemWorker. It creates workers on-demand for each new // ecosystem encountered. It also sends every vulnerability to the allEcosystemWorker.