Skip to content

fix(vulnfeeds): prevent OOM in combine-to-osv and optimize memory usage #6178

New issue

Have a question about this project? Sign up for a free GitHub account to open an issue and contact its maintainers and the community.

Sign up for GitHub

By clicking “Sign up for GitHub”, you agree to our terms of service and privacy statement. We’ll occasionally send you account related emails.

Already on GitHub? Sign in to your account

Merged
jess-lowe merged 2 commits into from
Oct 16, 2026
Merged

fix(vulnfeeds): prevent OOM in combine-to-osv and optimize memory usage #6178

Show file tree
Hide file tree
Changes from all commits
Commits
Show all changes
2 commits
Select commit Hold shift + click to select a range
b8ea73a
fix(vulnfeeds): prevent OOM in combine-to-osv by reducing writer chun…
jess-lowe Oct 16, 2026
9ac6cec
test(vulnfeeds): use protojson helper in combine-to-osv tests for dat…
jess-lowe Oct 16, 2026
File filter

Filter by extension

Filter by extension
Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
56 changes: 46 additions & 10 deletions vulnfeeds/cmd/combine-to-osv/main.go
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -263,7 +263,12 @@ func main() {
}
}

logger.Info("Total CVE Work Items to process", slog.Int("count", len(workItems)))
// Release cve5Files and nvdFiles slices to free memory
cve5Files = nil
nvdFiles = nil

totalWork := len(workItems)
logger.Info("Total CVE Work Items to process", slog.Int("count", totalWork))

// Start Upload Pool
var outBkt, overridesBkt *storage.BucketHandle
Expand Down Expand Up @@ -302,24 +307,28 @@ func main() {

// Interpose Collector to gather valid IDs
var validIDs []string
if *syncDeletions && *uploadToGCS {
validIDs = make([]string, 0, totalWork)
}
var validCount atomic.Uint64
var collectValidIDsWg sync.WaitGroup
collectValidIDsWg.Add(1)
totalWork := len(workItems)
go func() {
defer collectValidIDsWg.Done()
collectValidIDsWg.Go(func() {
count := 0
for v := range vulnChan {
count++
if len(v.GetAffected()) > 0 {
validIDs = append(validIDs, v.GetId())
if *syncDeletions && *uploadToGCS {
validIDs = append(validIDs, v.GetId())
}
validCount.Add(1)
}
uploadVulnsChan <- v
if count%1000 == 0 {
logger.Info("Processed CVEs", slog.Int("count", count), slog.Int("total", totalWork), slog.Int("percent", (count*100)/totalWork))
}
}
close(uploadVulnsChan)
}()
})

// Start ReadAndCombineWorkers (Read side)
var readAndCombineWg sync.WaitGroup
Expand All @@ -331,9 +340,11 @@ func main() {

// Feed Work
go func() {
for _, work := range workItems {
for id, work := range workItems {
workChan <- work
delete(workItems, id)
}
workItems = nil
close(workChan)
}()

Expand All @@ -345,7 +356,7 @@ func main() {
collectValidIDsWg.Wait()
uploadVulnsWg.Wait()

logger.Info("Successfully processed OSV files", slog.Int("count", len(validIDs)))
logger.Info("Successfully processed OSV files", slog.Uint64("count", validCount.Load()))
if outBkt == nil && gcsHelper == nil {
logger.Info("Successfully wrote records to disk", slog.Uint64("count", successCount.Load()))
}
Expand Down Expand Up @@ -822,7 +833,7 @@ func cleanLastAffectedIfFixedExists(r *osvschema.Range) {
}

// isReferencesOnly checks if the range 'source' field is only "REFERENCES"
// or ["REFERENCES"].
// or ["REFERENCES"], either in extracted_events or at the top level.
func isReferencesOnly(r *osvschema.Range) bool {
if r.GetDatabaseSpecific() == nil {
return false
Expand All @@ -831,6 +842,31 @@ func isReferencesOnly(r *osvschema.Range) bool {
if fields == nil {
return false
}

// Check new location inside extracted_events first
if extractedEventsVal, ok := fields["extracted_events"]; ok && extractedEventsVal.GetListValue() != nil {
groups := extractedEventsVal.GetListValue().GetValues()
if len(groups) > 0 {
allRefs := true
for _, groupVal := range groups {
groupS := groupVal.GetStructValue()
if groupS == nil {
allRefs = false
break
}
groupFields := groupS.GetFields()
sourceVal, ok := groupFields["source"]
if !ok || sourceVal.GetStringValue() != "REFERENCES" {
allRefs = false
break
}
}
if allRefs {
return true
}
}
}

val, ok := fields["source"]
if !ok {
return false
Expand Down
146 changes: 77 additions & 69 deletions vulnfeeds/cmd/combine-to-osv/main_test.go
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -8,11 +8,22 @@ import (
"github.com/google/go-cmp/cmp"
"github.com/google/go-cmp/cmp/cmpopts"
"github.com/ossf/osv-schema/bindings/go/osvschema"
"google.golang.org/protobuf/encoding/protojson"
"google.golang.org/protobuf/testing/protocmp"
"google.golang.org/protobuf/types/known/structpb"
"google.golang.org/protobuf/types/known/timestamppb"
)

func mustParseDBSpecific(t *testing.T, jsonStr string) *structpb.Struct {
t.Helper()
var s structpb.Struct
if err := protojson.Unmarshal([]byte(jsonStr), &s); err != nil {
t.Fatalf("failed to unmarshal database_specific JSON: %v", err)
}

return &s
}

func TestPickAffectedInformation(t *testing.T) {
repoA := "https://example.com/repo/a"
repoB := "https://example.com/repo/b"
Expand Down Expand Up @@ -384,11 +395,7 @@ func TestPickAffectedInformation(t *testing.T) {
{Introduced: "0"},
{Fixed: "1.0.1"},
},
DatabaseSpecific: &structpb.Struct{
Fields: map[string]*structpb.Value{
"source": structpb.NewStringValue("CPE_RANGE"),
},
},
DatabaseSpecific: mustParseDBSpecific(t, `{"source": "CPE_RANGE"}`),
},
},
},
Expand All @@ -403,11 +410,7 @@ func TestPickAffectedInformation(t *testing.T) {
{Introduced: "0"},
{Fixed: "1.0.1"},
},
DatabaseSpecific: &structpb.Struct{
Fields: map[string]*structpb.Value{
"source": structpb.NewStringValue("CPE_RANGE"),
},
},
DatabaseSpecific: mustParseDBSpecific(t, `{"source": "CPE_RANGE"}`),
},
},
},
Expand Down Expand Up @@ -439,16 +442,7 @@ func TestPickAffectedInformation(t *testing.T) {
{Introduced: "0"},
{Fixed: "1.0.1"},
},
DatabaseSpecific: &structpb.Struct{
Fields: map[string]*structpb.Value{
"source": structpb.NewListValue(&structpb.ListValue{
Values: []*structpb.Value{
structpb.NewStringValue("CPE_RANGE"),
structpb.NewStringValue("REFERENCES"),
},
}),
},
},
DatabaseSpecific: mustParseDBSpecific(t, `{"source": ["CPE_RANGE", "REFERENCES"]}`),
},
},
},
Expand All @@ -463,16 +457,7 @@ func TestPickAffectedInformation(t *testing.T) {
{Introduced: "0"},
{Fixed: "1.0.1"},
},
DatabaseSpecific: &structpb.Struct{
Fields: map[string]*structpb.Value{
"source": structpb.NewListValue(&structpb.ListValue{
Values: []*structpb.Value{
structpb.NewStringValue("CPE_RANGE"),
structpb.NewStringValue("REFERENCES"),
},
}),
},
},
DatabaseSpecific: mustParseDBSpecific(t, `{"source": ["CPE_RANGE", "REFERENCES"]}`),
},
},
},
Expand Down Expand Up @@ -521,11 +506,7 @@ func TestPickAffectedInformation(t *testing.T) {
{Introduced: "1.0.0"},
{LastAffected: "1.0.1"},
},
DatabaseSpecific: &structpb.Struct{
Fields: map[string]*structpb.Value{
"source": structpb.NewStringValue("AFFECTED_FIELD"),
},
},
DatabaseSpecific: mustParseDBSpecific(t, `{"source": "AFFECTED_FIELD"}`),
},
},
},
Expand All @@ -540,11 +521,7 @@ func TestPickAffectedInformation(t *testing.T) {
{Introduced: "0"},
{Fixed: "2c1762b85acb"},
},
DatabaseSpecific: &structpb.Struct{
Fields: map[string]*structpb.Value{
"source": structpb.NewStringValue("REFERENCES"),
},
},
DatabaseSpecific: mustParseDBSpecific(t, `{"source": "REFERENCES"}`),
},
},
},
Expand All @@ -559,16 +536,64 @@ func TestPickAffectedInformation(t *testing.T) {
{Introduced: "1.0.0"},
{Fixed: "2c1762b85acb"},
},
DatabaseSpecific: &structpb.Struct{
Fields: map[string]*structpb.Value{
"source": structpb.NewListValue(&structpb.ListValue{
Values: []*structpb.Value{
structpb.NewStringValue("AFFECTED_FIELD"),
structpb.NewStringValue("REFERENCES"),
},
}),
},
DatabaseSpecific: mustParseDBSpecific(t, `{"source": ["AFFECTED_FIELD", "REFERENCES"]}`),
},
},
},
},
},
{
name: "Merge references-only range with extracted_events source with CVE range",
cve5Affected: []*osvschema.Affected{
{
Ranges: []*osvschema.Range{
{
Type: osvschema.Range_GIT,
Repo: repoA,
Events: []*osvschema.Event{
{Introduced: "1.0.0"},
{LastAffected: "1.0.1"},
},
DatabaseSpecific: mustParseDBSpecific(t, `{"source": "AFFECTED_FIELD"}`),
},
},
},
},
nvdAffected: []*osvschema.Affected{
{
Ranges: []*osvschema.Range{
{
Type: osvschema.Range_GIT,
Repo: repoA,
Events: []*osvschema.Event{
{Introduced: "0"},
{Fixed: "2c1762b85acb"},
},
DatabaseSpecific: mustParseDBSpecific(t, `{
"extracted_events": [
{"source": "REFERENCES"}
]
}`),
},
},
},
},
wantAffected: []*osvschema.Affected{
{
Ranges: []*osvschema.Range{
{
Type: osvschema.Range_GIT,
Repo: repoA,
Events: []*osvschema.Event{
{Introduced: "1.0.0"},
{Fixed: "2c1762b85acb"},
},
DatabaseSpecific: mustParseDBSpecific(t, `{
"source": "AFFECTED_FIELD",
"extracted_events": [
{"source": "REFERENCES"}
]
}`),
},
},
},
Expand All @@ -586,11 +611,7 @@ func TestPickAffectedInformation(t *testing.T) {
{Introduced: "0"},
{Fixed: "e4dd3fa3182d0fd382e229e0c25d1bfd8b77a711"},
},
DatabaseSpecific: &structpb.Struct{
Fields: map[string]*structpb.Value{
"source": structpb.NewStringValue("AFFECTED_FIELD"),
},
},
DatabaseSpecific: mustParseDBSpecific(t, `{"source": "AFFECTED_FIELD"}`),
},
},
},
Expand All @@ -604,11 +625,7 @@ func TestPickAffectedInformation(t *testing.T) {
Events: []*osvschema.Event{
{Fixed: "83b3e91e0c1e84873a6d3ca3c5887eb5b4f5a3d8"},
},
DatabaseSpecific: &structpb.Struct{
Fields: map[string]*structpb.Value{
"source": structpb.NewStringValue("REFERENCES"),
},
},
DatabaseSpecific: mustParseDBSpecific(t, `{"source": "REFERENCES"}`),
},
},
},
Expand All @@ -623,16 +640,7 @@ func TestPickAffectedInformation(t *testing.T) {
{Introduced: "0"},
{Fixed: "83b3e91e0c1e84873a6d3ca3c5887eb5b4f5a3d8"},
},
DatabaseSpecific: &structpb.Struct{
Fields: map[string]*structpb.Value{
"source": structpb.NewListValue(&structpb.ListValue{
Values: []*structpb.Value{
structpb.NewStringValue("AFFECTED_FIELD"),
structpb.NewStringValue("REFERENCES"),
},
}),
},
},
DatabaseSpecific: mustParseDBSpecific(t, `{"source": ["AFFECTED_FIELD", "REFERENCES"]}`),
},
},
},
Expand Down
2 changes: 2 additions & 0 deletions vulnfeeds/conversion/writer/writer.go
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -18,6 +18,7 @@ import (
"time"

"cloud.google.com/go/storage"
"google.golang.org/api/googleapi"
"google.golang.org/protobuf/encoding/protojson"
"google.golang.org/protobuf/types/known/timestamppb"

Expand Down Expand Up @@ -103,6 +104,7 @@ func uploadIfChanged(ctx context.Context, v *osvschema.Vulnerability, hexHash st
logger.Info("Uploading vulnerability record to GCS", slog.String("id", vulnID), slog.String("object", objName))

wc := obj.NewWriter(ctx)
wc.ChunkSize = googleapi.MinUploadChunkSize

@michaelkedar michaelkedar Oct 16, 2026

Copy link
Copy Markdown
Member

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

this might slow things down considerably?

wc.Metadata = map[string]string{
hashMetadataKey: hexHash,
}
Expand Down
2 changes: 2 additions & 0 deletions vulnfeeds/gcs-tools/gcs.go
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -17,6 +17,7 @@ import (
"cloud.google.com/go/storage"
"github.com/google/osv.dev/vulnfeeds/utility/logger"
"golang.org/x/sync/errgroup"
"google.golang.org/api/googleapi"
"google.golang.org/api/iterator"
)

Expand Down Expand Up @@ -123,6 +124,7 @@ func (g *Helper) CloseAndWait() {
func UploadToGCS(ctx context.Context, bkt *storage.BucketHandle, objectName string, data io.Reader, contentType string, metadata map[string]string) error {
obj := bkt.Object(objectName)
wc := obj.NewWriter(ctx)
wc.ChunkSize = googleapi.MinUploadChunkSize
if contentType != "" {
wc.ContentType = contentType
}
Expand Down
Loading