From 73b9cf4fbefc9e1f7d27567581d387ffb7754c34 Mon Sep 17 00:00:00 2001 From: Matthew Staebler Date: Mon, 13 Jul 2026 21:44:04 -0400 Subject: [PATCH 1/2] Optimize prow loader with bulk writer, COPY protocol, and GCS improvements Restructure the prow loader from 50 concurrent goroutines each doing their own DB work into a producer/consumer pipeline: 50 GCS fetch workers feed results through a channel to a single bulk writer that batches and commits atomically using PostgreSQL COPY protocol and temp tables. Key changes: - Replace per-row GORM operations with COPY + temp table + single transaction per batch (100 job runs). ProwJobRun, annotations, pull requests, and tests all commit atomically. - Pre-filter already-loaded job runs via anti-join against prow_job_runs before GCS fetching, eliminating redundant work on steady-state loads. - Replace client-side regex filtering of GCS objects with server-side MatchGlob. - Pre-compile release config regexes at construction time. - Batch ProwJob upserts (100 per batch) instead of per-row. - Filter pending/triggered jobs in BQ query. - Remove per-row locks and caches that are no longer needed. Benchmarks (staging, release 4.19, 1h lookback): - Cold load: 5.1x faster (108s -> 21s) - 48h steady-state: 43s (13,604 candidates, 72 new jobs) Co-Authored-By: Claude Opus 4.6 --- pkg/api/jobrunevents/job_run_events.go | 10 +- pkg/api/jobrunintervals/job_run_intervals.go | 12 +- pkg/dataloader/prowloader/bigqueryjobs.go | 1 + .../prowloader/extract_test_cases_test.go | 124 ++ pkg/dataloader/prowloader/gcs/gcs_jobrun.go | 95 +- pkg/dataloader/prowloader/prow.go | 1018 +++++++++++------ pkg/db/suites.go | 30 +- pkg/variantregistry/ocp.go | 8 +- 8 files changed, 813 insertions(+), 485 deletions(-) create mode 100644 pkg/dataloader/prowloader/extract_test_cases_test.go diff --git a/pkg/api/jobrunevents/job_run_events.go b/pkg/api/jobrunevents/job_run_events.go index c77c766b0d..5a4f026c66 100644 --- a/pkg/api/jobrunevents/job_run_events.go +++ b/pkg/api/jobrunevents/job_run_events.go @@ -5,7 +5,6 @@ import ( "encoding/json" "errors" "fmt" - "regexp" "strings" "cloud.google.com/go/storage" @@ -16,9 +15,6 @@ import ( "github.com/openshift/sippy/pkg/db" ) -// eventsJSONRegex matches paths like artifacts/*e2e*/gather-extra/artifacts/events.json -var eventsJSONRegex = regexp.MustCompile(`gather-extra/artifacts/events\.json$`) - // KubeEvent represents a flattened Kubernetes Event for the API response type KubeEvent struct { FirstTimestamp string `json:"firstTimestamp"` @@ -76,17 +72,17 @@ func JobRunEvents(gcsClient *storage.Client, dbc *db.DB, jobRunID int64, gcsBuck } gcsJobRun := gcs.NewGCSJobRun(gcsClient.Bucket(gcsBucket), gcsPath) - matches, err := gcsJobRun.FindAllMatches([]*regexp.Regexp{eventsJSONRegex}) + matches, err := gcsJobRun.FindAllMatches(context.TODO(), gcs.GlobEventsJSON) if err != nil { return &EventListResponse{JobRunURL: jobRunURL}, err } - if len(matches) == 0 || len(matches[0]) == 0 { + if len(matches) == 0 { logger.Info("no events.json file found") return &EventListResponse{Items: []KubeEvent{}, JobRunURL: jobRunURL}, nil } - eventsPath := matches[0][0] + eventsPath := matches[0] logger.WithField("events_path", eventsPath).Info("found events.json") content, err := gcsJobRun.GetContent(context.TODO(), eventsPath) diff --git a/pkg/api/jobrunintervals/job_run_intervals.go b/pkg/api/jobrunintervals/job_run_intervals.go index c834c5b262..64597f1088 100644 --- a/pkg/api/jobrunintervals/job_run_intervals.go +++ b/pkg/api/jobrunintervals/job_run_intervals.go @@ -5,7 +5,6 @@ import ( "encoding/json" "errors" "fmt" - "regexp" "strings" "cloud.google.com/go/storage" @@ -45,10 +44,15 @@ func JobRunIntervals(gcsClient *storage.Client, dbc *db.DB, jobRunID int64, gcsB gcsJobRun := gcs.NewGCSJobRun(gcsClient.Bucket(gcsBucket), gcsPath) intervals := &apitype.EventIntervalList{} - intervalFiles, err := gcsJobRun.FindAllMatches([]*regexp.Regexp{gcs.GetIntervalFile()}) + intervalFiles, err := gcsJobRun.FindAllMatches(context.TODO(), gcs.GlobIntervalsJSON) if err != nil { return intervals, err } + timelineFiles, err := gcsJobRun.FindAllMatches(context.TODO(), gcs.GlobTimelinesJSON) + if err != nil { + return intervals, err + } + intervalFiles = append(intervalFiles, timelineFiles...) // We will often match multiple files here, one for upgrade phase, one for conformance // testing phase. For now, we return them all, and each interval has a filename it @@ -57,7 +61,7 @@ func JobRunIntervals(gcsClient *storage.Client, dbc *db.DB, jobRunID int64, gcsB logger.Info("no interval files found") return intervals, nil } - logger.WithField("interval_files", intervalFiles[0]).Info("found interval files") + logger.WithField("interval_files", intervalFiles).Info("found interval files") // Now that we have the list of all matching interval files, if the user specified one, return only // intervals from that file. If they didn't, make a best guess on the default to load, our minimal @@ -66,7 +70,7 @@ func JobRunIntervals(gcsClient *storage.Client, dbc *db.DB, jobRunID int64, gcsB intervalFilesAvailable := []string{} // Find the full path to the filename specified: - for _, fp := range intervalFiles[0] { + for _, fp := range intervalFiles { // Get the base filename we'll add to each incoming interval: tokens := strings.Split(fp, "/") baseFile := tokens[len(tokens)-1] diff --git a/pkg/dataloader/prowloader/bigqueryjobs.go b/pkg/dataloader/prowloader/bigqueryjobs.go index f818176693..8730a201e4 100644 --- a/pkg/dataloader/prowloader/bigqueryjobs.go +++ b/pkg/dataloader/prowloader/bigqueryjobs.go @@ -64,6 +64,7 @@ func (pl *ProwLoader) fetchProwJobsFromOpenShiftBigQuery() ([]prow.ProwJob, []er "FROM `ci_analysis_us.jobs` "+ `WHERE TIMESTAMP(prowjob_completion) > @queryFrom AND prowjob_url IS NOT NULL + AND prowjob_state NOT IN ('pending', 'triggered') ORDER BY prowjob_start_ts`) query.Parameters = []bigquery.QueryParameter{ { diff --git a/pkg/dataloader/prowloader/extract_test_cases_test.go b/pkg/dataloader/prowloader/extract_test_cases_test.go new file mode 100644 index 0000000000..c6e801e64f --- /dev/null +++ b/pkg/dataloader/prowloader/extract_test_cases_test.go @@ -0,0 +1,124 @@ +package prowloader + +import ( + "testing" + + "github.com/stretchr/testify/assert" + + "github.com/openshift/sippy/pkg/apis/junit" + sippyprocessingv1 "github.com/openshift/sippy/pkg/apis/sippyprocessing/v1" +) + +func TestExtractTestCases(t *testing.T) { + failMsg := "something went wrong" + + tests := []struct { + name string + suite *junit.TestSuite + expected map[string]*testCaseEntry + }{ + { + name: "passing test", + suite: &junit.TestSuite{ + Name: "openshift-tests", + TestCases: []*junit.TestCase{ + {Name: "test-a", Duration: 1.5}, + }, + }, + expected: map[string]*testCaseEntry{ + "openshift-tests.test-a": { + TestName: "test-a", + SuiteName: "openshift-tests", + Status: int(sippyprocessingv1.TestStatusSuccess), + Duration: 1.5, + }, + }, + }, + { + name: "failing test with output", + suite: &junit.TestSuite{ + Name: "openshift-tests", + TestCases: []*junit.TestCase{ + {Name: "test-a", Duration: 2.0, FailureOutput: &junit.FailureOutput{Output: failMsg}}, + }, + }, + expected: map[string]*testCaseEntry{ + "openshift-tests.test-a": { + TestName: "test-a", + SuiteName: "openshift-tests", + Status: int(sippyprocessingv1.TestStatusFailure), + Duration: 2.0, + Output: &failMsg, + }, + }, + }, + { + name: "skipped test excluded", + suite: &junit.TestSuite{ + Name: "openshift-tests", + TestCases: []*junit.TestCase{ + {Name: "test-a", SkipMessage: &junit.SkipMessage{Message: "skipped"}}, + }, + }, + expected: map[string]*testCaseEntry{}, + }, + { + name: "flake from pass then fail", + suite: &junit.TestSuite{ + Name: "openshift-tests", + TestCases: []*junit.TestCase{ + {Name: "test-a", Duration: 1.0}, + {Name: "test-a", Duration: 2.0, FailureOutput: &junit.FailureOutput{Output: failMsg}}, + }, + }, + expected: map[string]*testCaseEntry{ + "openshift-tests.test-a": { + TestName: "test-a", + SuiteName: "openshift-tests", + Status: int(sippyprocessingv1.TestStatusFlake), + Duration: 1.0, + Output: &failMsg, + }, + }, + }, + { + name: "child suite uses its own name", + suite: &junit.TestSuite{ + Name: "openshift-tests", + TestCases: []*junit.TestCase{ + {Name: "parent-test", Duration: 1.0}, + }, + Children: []*junit.TestSuite{ + { + Name: "k8s.io", + TestCases: []*junit.TestCase{ + {Name: "child-test", Duration: 3.0}, + }, + }, + }, + }, + expected: map[string]*testCaseEntry{ + "openshift-tests.parent-test": { + TestName: "parent-test", + SuiteName: "openshift-tests", + Status: int(sippyprocessingv1.TestStatusSuccess), + Duration: 1.0, + }, + "k8s.io.child-test": { + TestName: "child-test", + SuiteName: "k8s.io", + Status: int(sippyprocessingv1.TestStatusSuccess), + Duration: 3.0, + }, + }, + }, + } + + for _, tt := range tests { + t.Run(tt.name, func(t *testing.T) { + testCases := make(map[string]*testCaseEntry) + extractTestCases(tt.suite, testCases) + assert.Equal(t, tt.expected, testCases) + }) + } +} diff --git a/pkg/dataloader/prowloader/gcs/gcs_jobrun.go b/pkg/dataloader/prowloader/gcs/gcs_jobrun.go index 79550c2128..7305563155 100644 --- a/pkg/dataloader/prowloader/gcs/gcs_jobrun.go +++ b/pkg/dataloader/prowloader/gcs/gcs_jobrun.go @@ -16,17 +16,17 @@ import ( ) const TestFailureSummaryFilePrefix = "risk-analysis" -const ClusterDataFilePrefix = "cluster-data_" -const JunitRegExStr = "\\/.*junit.*xml" -const intervalFilesRegExStr = "\\/(e2e-events|e2e-timelines).*json" - -var ( - defaultRiskAnalysisSummaryFileRegEx *regexp.Regexp - defaultClusterDataFileRegEx *regexp.Regexp - defaultJunitFileRegEx *regexp.Regexp - intervalFilesRegex *regexp.Regexp + +const ( + GlobJunitXML = "**junit*.xml" + GlobEventsJSON = "**/gather-extra/artifacts/events.json" + GlobIntervalsJSON = "**e2e-events*.json" + GlobTimelinesJSON = "**e2e-timelines*.json" + GlobClusterData = "**/cluster-data_*.json" ) +var defaultRiskAnalysisSummaryFileRegEx *regexp.Regexp + func GetDefaultRiskAnalysisSummaryFile() *regexp.Regexp { if defaultRiskAnalysisSummaryFileRegEx == nil { defaultRiskAnalysisSummaryFileRegEx = regexp.MustCompile(fmt.Sprintf("%s.json", TestFailureSummaryFilePrefix)) @@ -34,27 +34,6 @@ func GetDefaultRiskAnalysisSummaryFile() *regexp.Regexp { return defaultRiskAnalysisSummaryFileRegEx } -func GetDefaultClusterDataFile() *regexp.Regexp { - if defaultClusterDataFileRegEx == nil { - defaultClusterDataFileRegEx = regexp.MustCompile(fmt.Sprintf("%s.*json", ClusterDataFilePrefix)) - } - return defaultClusterDataFileRegEx -} - -func GetDefaultJunitFile() *regexp.Regexp { - if defaultJunitFileRegEx == nil { - defaultJunitFileRegEx = regexp.MustCompile(JunitRegExStr) - } - return defaultJunitFileRegEx -} - -func GetIntervalFile() *regexp.Regexp { - if intervalFilesRegex == nil { - intervalFilesRegex = regexp.MustCompile(intervalFilesRegExStr) - } - return intervalFilesRegex -} - type GCSJobRun struct { // retrieval mechanisms bkt *storage.BucketHandle @@ -76,16 +55,13 @@ func (j *GCSJobRun) SetGCSJunitPaths(paths []string) { j.gcsJunitPaths = paths } -func (j *GCSJobRun) GetGCSJunitPaths() ([]string, error) { +func (j *GCSJobRun) GetGCSJunitPaths(ctx context.Context) ([]string, error) { if len(j.gcsJunitPaths) == 0 { - matches, err := j.FindAllMatches([]*regexp.Regexp{GetDefaultJunitFile()}) + matches, err := j.FindAllMatches(ctx, GlobJunitXML) if err != nil { return nil, err } - - if len(matches) > 0 { - j.gcsJunitPaths = matches[0] - } + j.gcsJunitPaths = matches } return j.gcsJunitPaths, nil @@ -93,7 +69,7 @@ func (j *GCSJobRun) GetGCSJunitPaths() ([]string, error) { func (j *GCSJobRun) GetCombinedJUnitTestSuites(ctx context.Context) (*junit.TestSuites, error) { testSuites := &junit.TestSuites{} - junitPaths, err := j.GetGCSJunitPaths() + junitPaths, err := j.GetGCSJunitPaths(ctx) if err != nil { return nil, err } @@ -135,19 +111,7 @@ func (j *GCSJobRun) GetContent(ctx context.Context, path string) ([]byte, error) return content, nil } - // Get an Object handle for the path obj := j.bkt.Object(path) - - // use the object attributes to try to get the latest generation to try to retrieve the data without getting a cached - // version of data that does not match the latest content. I don't know if this will work, but in the easy case - // it doesn't seem to fail. - objAttrs, err := obj.Attrs(ctx) - if err != nil { - return nil, fmt.Errorf("error reading GCS attributes for jobrun: %w", err) - } - obj = obj.Generation(objAttrs.Generation) - - // Get an io.Reader for the object. gcsReader, err := obj.NewReader(ctx) if err != nil { return nil, fmt.Errorf("error reading GCS content for jobrun: %w", err) @@ -196,20 +160,19 @@ func (j *GCSJobRun) FindFirstFile(root string, filename *regexp.Regexp) []byte { return nil } -// FindAllMatches takes an array of regexes -// and compares the name of the object in gcs -// with each regex for a match -// each regex that matches will get the attribute name -// in the returned matches with the index matching the regex -func (j *GCSJobRun) FindAllMatches(filenames []*regexp.Regexp) ([][]string, error) { - if len(filenames) < 1 { - return nil, nil +// FindAllMatches lists GCS objects under the job run path that match the +// given glob pattern, using server-side filtering via MatchGlob. +func (j *GCSJobRun) FindAllMatches(ctx context.Context, glob string) ([]string, error) { + q := &storage.Query{ + Prefix: j.gcsProwJobPath, + MatchGlob: glob, + } + if err := q.SetAttrSelection([]string{"Name"}); err != nil { + return nil, errors.Wrap(err, "error setting attribute selection") } - matches := make([][]string, len(filenames)) - it := j.bkt.Objects(context.Background(), &storage.Query{ - Prefix: j.gcsProwJobPath, - }) + var matches []string + it := j.bkt.Objects(ctx, q) for { attrs, err := it.Next() if errors.Is(err, iterator.Done) { @@ -218,15 +181,7 @@ func (j *GCSJobRun) FindAllMatches(filenames []*regexp.Regexp) ([][]string, erro if err != nil { return nil, errors.Wrap(err, "error reading GCS attributes for job run") } - - for i, filename := range filenames { - if matches[i] == nil { - matches[i] = make([]string, 0) - } - if filename.MatchString(attrs.Name) { - matches[i] = append(matches[i], attrs.Name) - } - } + matches = append(matches, attrs.Name) } return matches, nil diff --git a/pkg/dataloader/prowloader/prow.go b/pkg/dataloader/prowloader/prow.go index ab2d09e169..b4cf2ddb2a 100644 --- a/pkg/dataloader/prowloader/prow.go +++ b/pkg/dataloader/prowloader/prow.go @@ -9,7 +9,6 @@ import ( "net/http" "net/url" "os" - "reflect" "regexp" "strconv" "sync" @@ -21,6 +20,7 @@ import ( "cloud.google.com/go/storage" "github.com/jackc/pgtype" "github.com/jackc/pgx/v4" + "github.com/jackc/pgx/v4/stdlib" "github.com/lib/pq" "github.com/openshift/sippy/pkg/bigquery/bqlabel" "github.com/pkg/errors" @@ -64,18 +64,12 @@ type ProwLoader struct { bigQueryClient *bqcachedclient.Client maxConcurrency int prowJobCache map[string]*models.ProwJob - prowJobCacheLock sync.RWMutex - prowJobRunCache map[uint]bool - prowJobRunCacheLock sync.RWMutex - prowJobRunTestCache map[string]uint - prowJobRunTestCacheLock sync.RWMutex variantManager testidentification.VariantManager - suiteCache map[string]*uint - suiteCacheLock sync.RWMutex syntheticTestManager synthetictests.SyntheticTestManager syntheticReleaseJobOverrides *releaseoverride.SyntheticReleaseOverrides releases []string - releaseSet map[string]bool + releaseSet sets.Set[string] + releaseRegexps map[string][]*regexp.Regexp config *v1config.SippyConfig ghCommenter *commenter.GitHubCommenter jobsImportedCount atomic.Int32 @@ -101,22 +95,33 @@ func New( loadSince *time.Time, syntheticReleaseJobOverrides *releaseoverride.SyntheticReleaseOverrides) *ProwLoader { + compiledRegexps := make(map[string][]*regexp.Regexp, len(releases)) + for _, release := range releases { + if cfg, ok := config.Releases[release]; ok { + for _, expr := range cfg.Regexp { + re, err := regexp.Compile(expr) + if err != nil { + log.WithError(err).WithField("release", release).WithField("regex", expr).Error("invalid regex in configuration") + continue + } + compiledRegexps[release] = append(compiledRegexps[release], re) + } + } + } + return &ProwLoader{ ctx: ctx, dbc: dbc, gcsClient: gcsClient, githubClient: githubClient, bigQueryClient: bigQueryClient, - maxConcurrency: 10, - prowJobRunCache: loadProwJobRunCache(dbc), - prowJobCache: loadProwJobCache(dbc), - prowJobRunTestCache: make(map[string]uint), - suiteCache: make(map[string]*uint), + maxConcurrency: 50, syntheticTestManager: syntheticTestManager, syntheticReleaseJobOverrides: syntheticReleaseJobOverrides, variantManager: variantManager, releases: releases, - releaseSet: toSet(releases), + releaseSet: sets.New[string](releases...), + releaseRegexps: compiledRegexps, config: config, ghCommenter: ghCommenter, promPusher: promPusher, @@ -137,14 +142,6 @@ func (pl *ProwLoader) resolveLoadSince() time.Time { return resolveFrom(pl.loadSince, time.Now()) } -func toSet(items []string) map[string]bool { - s := make(map[string]bool, len(items)) - for _, item := range items { - s[item] = true - } - return s -} - var clusterDataDateTimeName = regexp.MustCompile(`cluster-data_(?P.*)-(?P