Optimize prow loader with bulk writer, COPY protocol, and GCS improvements#3768
Conversation
|
Pipeline controller notification For optional jobs, comment This repository is configured in: automatic mode |
openshift-ci
Bot
added
the
do-not-merge/work-in-progress
|
Skipping CI for Draft Pull Request. |
|
@coderabbitai full review |
✅ Action performedFull review finished. |
|
No actionable comments were generated in the recent review. 🎉 ℹ️ Recent review info⚙️ Run configurationConfiguration used: Repository YAML (base), Central YAML (inherited) Review profile: CHILL Plan: Enterprise Run ID: 📒 Files selected for processing (2)
🚧 Files skipped from review as they are similar to previous changes (2)
WalkthroughThe PR replaces regex-based GCS file discovery with glob queries, excludes pending and triggered BigQuery jobs, and restructures Prow ingestion into preprocessing, concurrent job-run fetching, and batched database persistence with expanded test coverage. ChangesProw ingestion and GCS discovery
Estimated code review effort: 4 (Complex) | ~60 minutes Sequence Diagram(s)sequenceDiagram
participant BigQuery
participant ProwLoader
participant GCSJobRun
participant PostgreSQL
BigQuery->>ProwLoader: Load eligible non-pending, non-triggered jobs
ProwLoader->>PostgreSQL: Bulk upsert ProwJob definitions
ProwLoader->>GCSJobRun: Fetch JUnit and job-run artifacts
GCSJobRun-->>ProwLoader: Return jobRunResult rows
ProwLoader->>PostgreSQL: Write job-run batches in a transaction
Suggested labels: Suggested reviewers: 🚥 Pre-merge checks | ✅ 18 | ❌ 3❌ Failed checks (3 warnings)
✅ Passed checks (18 passed)
✨ Finishing Touches🧪 Generate unit tests (beta)
Thanks for using CodeRabbit! It's free for OSS, and your support helps us grow. If you like it, consider giving us a shout-out. Comment |
![coderabbitai[bot]](https://avatars.githubusercontent.com/in/347564?s=60&v=4){kind=small}
There was a problem hiding this comment.
Actionable comments posted: 4
Caution
Some comments are outside the diff and can’t be posted inline due to platform limitations.
⚠️ Outside diff range comments (1)
pkg/dataloader/prowloader/prow.go (1)
163-172: 🩺 Stability & Availability | 🟠 Major | ⚡ Quick winSilent cache-population failure:
loadProwJobCachedrops the GORM query error entirely.
dbc.DB.Model(&models.ProwJob{}).Find(&allJobs)'s returned*gorm.DB(and its.Error) is never captured. Previously this ran once at startup; nowpreprocessProwJobsre-runs it everyLoad()cycle andpl.prowJobCachebecomes the sole source every one of the 50 concurrentfetchJobRunResultworkers relies on (dbProwJob, ok := pl.prowJobCache[pj.Spec.Job]). If this query transiently fails, the cache silently becomes empty, every job run is silently skipped with only a per-job warning log, andpl.errorsnever reflects the true root cause.As per path instructions,
**/*.go: "do not ignore returned errors with_without clear justification". Based on learnings, one-off context propagation isn't required here — just capture and propagate the existing error.🐛 Proposed fix
-func loadProwJobCache(dbc *db.DB) map[string]*models.ProwJob { +func loadProwJobCache(dbc *db.DB) (map[string]*models.ProwJob, error) { prowJobCache := map[string]*models.ProwJob{} var allJobs []*models.ProwJob - dbc.DB.Model(&models.ProwJob{}).Find(&allJobs) + if err := dbc.DB.Model(&models.ProwJob{}).Find(&allJobs).Error; err != nil { + return nil, fmt.Errorf("loading prow job cache: %w", err) + } for _, j := range allJobs { prowJobCache[j.Name] = j } log.Infof("job cache created with %d entries from database", len(prowJobCache)) - return prowJobCache + return prowJobCache, nil }- pl.prowJobCache = loadProwJobCache(pl.dbc) - return entries, nil + cache, err := loadProwJobCache(pl.dbc) + if err != nil { + return nil, fmt.Errorf("reloading prow job cache: %w", err) + } + pl.prowJobCache = cache + return entries, nilAlso applies to: 709-711
🤖 Prompt for AI Agents
Verify each finding against current code. Fix only still-valid issues, skip the rest with a brief reason, keep changes minimal, and validate. In `@pkg/dataloader/prowloader/prow.go` around lines 163 - 172, Update loadProwJobCache to capture the *gorm.DB result from the Find query and propagate its Error instead of returning only a cache map. Adjust callers such as preprocessProwJobs to handle the returned error and record it in the existing pl.errors flow, while preserving successful cache population and logging behavior.Source: Path instructions
🧹 Nitpick comments (4)
pkg/db/suites.go (1)
88-106: 🚀 Performance & Scalability | 🔵 Trivial | ⚡ Quick winConsider a
sets.Set[string]for the explicit-list membership check.
IsSuiteImportablelinearly scanstestSuites(~40 entries) on every call; this runs at least twice per job run across a 50-way-concurrent batch loader. A package-levelsets.Set[string]built once fromtestSuiteswould make this O(1) and aligns with the "usesets.Setfor unique-string collections" coding guideline already applied elsewhere in this PR (e.g.releaseSetinprow.go).♻️ Proposed fix
+var testSuiteSet = sets.New(testSuites...) + func IsSuiteImportable(name string) bool { - // Check explicit list - for _, s := range testSuites { - if s == name { - return true - } - } + if testSuiteSet.Has(name) { + return true + } // Check patterns for _, re := range testSuitePatterns {🤖 Prompt for AI Agents
Verify each finding against current code. Fix only still-valid issues, skip the rest with a brief reason, keep changes minimal, and validate. In `@pkg/db/suites.go` around lines 88 - 106, Update IsSuiteImportable to use a package-level sets.Set[string] initialized once from testSuites for explicit suite membership checks, replacing the per-call linear scan; retain the existing testSuitePatterns matching and return behavior unchanged.Source: Coding guidelines
pkg/dataloader/prowloader/prow.go (3)
293-310: 🚀 Performance & Scalability | 🔵 Trivial | ⚡ Quick winPer-job-run
Infofon a 50-way-concurrent hot path may add meaningful log volume/overhead at scale.
log.Infof("%d of %d job runs processed", ...)fires once per job run across up to 50 concurrent workers. For a cold load processing tens of thousands of runs, this is a very high volume of INFO logs competing for the shared logger/stdout, working against this PR's stated latency goals. Consider throttling (e.g., log every N runs or on a timer) instead of per-item.🤖 Prompt for AI Agents
Verify each finding against current code. Fix only still-valid issues, skip the rest with a brief reason, keep changes minimal, and validate. In `@pkg/dataloader/prowloader/prow.go` around lines 293 - 310, Throttle the progress logging in the worker loop around jobsImportedCount so it does not emit an Infof for every job run. Log progress only at a suitable interval or count threshold while preserving the existing counter updates and final processing behavior.
996-1013: 📐 Maintainability & Code Quality | 🔵 Trivial | ⚡ Quick winRedundant
BuildIDparsing acrosspreprocessProwJobsandfetchJobRunResult.
strconv.ParseUint(pj.Status.BuildID, 0, 64)is computed inpreprocessProwJobs(line 655) to buildcandidate.id/candidateIDs, but that value is discarded —entriesonly carries*prow.ProwJob.fetchJobRunResultthen re-parses the same string for every entry. As per path instructions,pkg/**/*.go: "Avoid calling the same utility function multiple times with identical arguments in the same code path." Threading the parsed ID throughentriesavoids the duplicate parse.Also applies to: 655-661
🤖 Prompt for AI Agents
Verify each finding against current code. Fix only still-valid issues, skip the rest with a brief reason, keep changes minimal, and validate. In `@pkg/dataloader/prowloader/prow.go` around lines 996 - 1013, Thread the parsed BuildID from preprocessProwJobs into each entry instead of discarding candidate.id, preserving it alongside the *prow.ProwJob. Update fetchJobRunResult and its callers to accept and use the parsed uint64 value, removing the redundant strconv.ParseUint call while retaining invalid-ID handling in preprocessProwJobs.Source: Path instructions
300-300: 📐 Maintainability & Code Quality | 🔵 Trivial | ⚡ Quick winPrefer structured fields for job/build IDs here, per coding guideline.
log.WithError(err).Warningf("couldn't fetch job %s/%s, continuing", pj.Spec.Job, pj.Status.BuildID)formats the job name and build ID into the message string. As per coding guidelines: "Prefer structured logging where it makes sense, particularly for names and IDs; often alog.WithField()call is preferred over formatting values into a string."♻️ Proposed fix
- fetchErrsCh <- err - log.WithError(err).Warningf("couldn't fetch job %s/%s, continuing", pj.Spec.Job, pj.Status.BuildID) + fetchErrsCh <- err + log.WithError(err).WithField("job", pj.Spec.Job).WithField("buildID", pj.Status.BuildID). + Warning("couldn't fetch job, continuing")🤖 Prompt for AI Agents
Verify each finding against current code. Fix only still-valid issues, skip the rest with a brief reason, keep changes minimal, and validate. In `@pkg/dataloader/prowloader/prow.go` at line 300, Update the warning log in the job-fetch error path to attach pj.Spec.Job and pj.Status.BuildID as structured fields via the existing log entry, while keeping the error and warning message intact without formatting those identifiers into it.Source: Coding guidelines
🤖 Prompt for all review comments with AI agents
Verify each finding against current code. Fix only still-valid issues, skip the
rest with a brief reason, keep changes minimal, and validate.
Inline comments:
In `@pkg/dataloader/prowloader/prow.go`:
- Around line 1645-1647: Replace the raw status value 12 in the failure-counting
logic with the named sippyprocessingv1.TestStatusFailure constant, matching the
existing constant usage in extractTestCases and preserving the current failures
increment behavior.
- Around line 1334-1344: Update the prow_pull_requests upsert in the surrounding
transaction to refresh author and title on conflicts, reusing the incoming
EXCLUDED values so previously blank fields can be backfilled while preserving
the existing merged_at and updated_at behavior. Confirm this matches the former
findOrAddPullRequests backfill semantics.
- Around line 1358-1396: Update the test upsert in the tests-existence SQL
within the surrounding loader flow so a name conflict with a soft-deleted row
clears deleted_at and restores the test instead of doing nothing. Preserve the
existing behavior for active tests and ensure the subsequent INNER JOIN tests
... deleted_at IS NULL can include reappearing tests; leave suite handling
unchanged.
- Line 1148: Convert the Labels value in the relevant prow job result
construction to a native []string before it reaches db.CopyToTempTable and pgx
CopyFrom. Follow the existing overall_result conversion pattern, while leaving
the labelsCache lookup and surrounding fields unchanged.
---
Outside diff comments:
In `@pkg/dataloader/prowloader/prow.go`:
- Around line 163-172: Update loadProwJobCache to capture the *gorm.DB result
from the Find query and propagate its Error instead of returning only a cache
map. Adjust callers such as preprocessProwJobs to handle the returned error and
record it in the existing pl.errors flow, while preserving successful cache
population and logging behavior.
---
Nitpick comments:
In `@pkg/dataloader/prowloader/prow.go`:
- Around line 293-310: Throttle the progress logging in the worker loop around
jobsImportedCount so it does not emit an Infof for every job run. Log progress
only at a suitable interval or count threshold while preserving the existing
counter updates and final processing behavior.
- Around line 996-1013: Thread the parsed BuildID from preprocessProwJobs into
each entry instead of discarding candidate.id, preserving it alongside the
*prow.ProwJob. Update fetchJobRunResult and its callers to accept and use the
parsed uint64 value, removing the redundant strconv.ParseUint call while
retaining invalid-ID handling in preprocessProwJobs.
- Line 300: Update the warning log in the job-fetch error path to attach
pj.Spec.Job and pj.Status.BuildID as structured fields via the existing log
entry, while keeping the error and warning message intact without formatting
those identifiers into it.
In `@pkg/db/suites.go`:
- Around line 88-106: Update IsSuiteImportable to use a package-level
sets.Set[string] initialized once from testSuites for explicit suite membership
checks, replacing the per-call linear scan; retain the existing
testSuitePatterns matching and return behavior unchanged.
🪄 Autofix (Beta)
Fix all unresolved CodeRabbit comments on this PR:
- Push a commit to this branch (recommended)
- Create a new PR with the fixes
ℹ️ Review info
⚙️ Run configuration
Configuration used: Repository YAML (base), Central YAML (inherited)
Review profile: CHILL
Plan: Enterprise
Run ID: e64f9dcc-1da5-48be-a9bb-7e10d19c3ab2
📒 Files selected for processing (8)
pkg/api/jobrunevents/job_run_events.gopkg/api/jobrunintervals/job_run_intervals.gopkg/dataloader/prowloader/bigqueryjobs.gopkg/dataloader/prowloader/extract_test_cases_test.gopkg/dataloader/prowloader/gcs/gcs_jobrun.gopkg/dataloader/prowloader/prow.gopkg/db/suites.gopkg/variantregistry/ocp.go
mstaeble
force-pushed
the
prow-loader-bulk-writer
branch
from
e59882a to
ed02f66
Compare
openshift-ci
Bot
added
the
ready-for-human-review
mstaeble
changed the title
openshift-ci
Bot
removed
the
do-not-merge/work-in-progress
|
Scheduling required tests: |
| defaultJunitFileRegEx *regexp.Regexp | ||
| intervalFilesRegex *regexp.Regexp | ||
| ) | ||
|
|
There was a problem hiding this comment.
What are your thoughts on keeping the globs defined as constants in a common group?
**junit*xml is used in multiple places and keeping the artifact list we pull in grouped together could make it a little easier to maintain / read.
Also noting that the . is getting dropped for the junit, not necessarily problematic but was it intentional?
There was a problem hiding this comment.
Good suggestions on both counts. Extracted the globs as named constants (GlobJunitXML, GlobEventsJSON, GlobIntervalsJSON, GlobTimelinesJSON, GlobClusterData) in gcs_jobrun.go and updated all callers. Also added the literal . before extensions in all globs (*junit.xml, *e2e-events.json, etc.). The missing dot was unintentional — the original regex (junit.*xml) didn't enforce it either, but it should have.
| } | ||
|
|
||
| runCleanup, err := db.CopyToTempTable(ctx, conn, "tmp_prow_job_runs", runs, runCols) | ||
| defer runCleanup() |
There was a problem hiding this comment.
code-review:pr prefers
if err != nil { return err} ; defer runCleanup()
multiple places below
There was a problem hiding this comment.
Agreed. Restructured all CopyToTempTable call sites to check the error before deferring cleanup:
if cleanup, err := db.CopyToTempTable(...); err != nil {
return err
} else {
defer cleanup()
}
There was a problem hiding this comment.
Well the linter doesn't like that, so instead it will be the following.
cleanup, err := ...
if err != nil {
return err
}
defer cleanup()
| {Name: "id", Type: "bigint NOT NULL", Value: func(id *uint) any { return *id }}, | ||
| }, | ||
| ) | ||
| defer cleanup() |
There was a problem hiding this comment.
code-review:pr prefers
if err != nil { return err} ; defer cleanup()
But concern is that db.CopyToTempTable might return nil cleanup on error which doesn't appear to be the case. Feel free to ignore.
There was a problem hiding this comment.
Same pattern applied here as well.
…ments 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 <noreply@anthropic.com>
mstaeble
force-pushed
the
prow-loader-bulk-writer
branch
from
ed02f66 to
73b9cf4
Compare
There was a problem hiding this comment.
Actionable comments posted: 3
🤖 Prompt for all review comments with AI agents
Verify each finding against current code. Fix only still-valid issues, skip the
rest with a brief reason, keep changes minimal, and validate.
Inline comments:
In `@pkg/dataloader/prowloader/prow.go`:
- Around line 658-664: Update both build-ID parsing sites in the
candidate-loading flow, including the block around candidate creation and the
corresponding logic near the second occurrence, to parse decimal values with a
63-bit limit rather than base-0 unsigned 64-bit parsing. Continue skipping
candidates whenever parsing fails or the value exceeds PostgreSQL bigint’s
positive range, and preserve the existing candidate append behavior for valid
IDs.
- Around line 749-753: Update the candidate-ID existence check around the query
used by the insert flow to treat tombstoned prow_job_runs records as existing,
preventing them from reaching the plain INSERT path and causing batch rollback.
Apply the same correction to the corresponding logic near the second referenced
location, and preserve the intended deleted-versus-restored behavior
consistently for dependent records.
- Line 1618: Update the test identity map in the surrounding test-processing
logic to use a comparable struct containing separate suite and test fields,
preventing collisions from dots in either name. Keep the legacy
synthetic-conversion key separate from this structured identity, and add a
regression test covering distinct dotted suite/test names such as (“a.b”, “c”)
and (“a”, “b.c”).
🪄 Autofix (Beta)
Fix all unresolved CodeRabbit comments on this PR:
- Push a commit to this branch (recommended)
- Create a new PR with the fixes
ℹ️ Review info
⚙️ Run configuration
Configuration used: Repository YAML (base), Central YAML (inherited)
Review profile: CHILL
Plan: Enterprise
Run ID: 1f55240a-1ee2-4e27-887d-0127af25a360
📒 Files selected for processing (8)
pkg/api/jobrunevents/job_run_events.gopkg/api/jobrunintervals/job_run_intervals.gopkg/dataloader/prowloader/bigqueryjobs.gopkg/dataloader/prowloader/extract_test_cases_test.gopkg/dataloader/prowloader/gcs/gcs_jobrun.gopkg/dataloader/prowloader/prow.gopkg/db/suites.gopkg/variantregistry/ocp.go
🚧 Files skipped from review as they are similar to previous changes (6)
- pkg/dataloader/prowloader/extract_test_cases_test.go
- pkg/api/jobrunintervals/job_run_intervals.go
- pkg/api/jobrunevents/job_run_events.go
- pkg/dataloader/prowloader/gcs/gcs_jobrun.go
- pkg/db/suites.go
- pkg/variantregistry/ocp.go
…for test dedup - ParseUint: use base 10 and 63-bit limit to match PostgreSQL bigint range - Anti-join: remove deleted_at filter so soft-deleted runs are not re-inserted - Replace dot-delimited string key with testCaseKey struct to prevent collisions - Add regression test for dotted suite/test name combinations Co-Authored-By: Claude Opus 4.6 <noreply@anthropic.com>
|
Scheduling required tests: |
|
@mstaeble: all tests passed! Full PR test history. Your PR dashboard. DetailsInstructions for interacting with me using PR comments are available here. If you have questions or suggestions related to my behavior, please file an issue against the kubernetes-sigs/prow repository. I understand the commands that are listed here. |
|
/lgtm |
|
[APPROVALNOTIFIER] This PR is APPROVED This pull-request has been approved by: mstaeble, neisw The full list of commands accepted by this bot can be found here. The pull request process is described here DetailsNeeds approval from an approver in each of these files:
Approvers can indicate their approval by writing |
openshift-ci
Bot
added
the
approved
2 participants
Summary
Restructures 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.
MatchGlobfor allFindAllMatchescallers.pending/triggeredjobs at the query level.Benchmarks (staging, release 4.19, 1h lookback)
Parity test confirmed: identical run IDs, test counts, and annotations between main and bulk-writer for the same BQ window.
Test plan
make lintpassesmake testpassesmake e2epasses🤖 Generated with Claude Code
Summary by CodeRabbit
events.jsonis missing.