diff --git a/pkg/api/componentreadiness/dataprovider/postgres/provider.go b/pkg/api/componentreadiness/dataprovider/postgres/provider.go index d762274bc..b88ac1f4d 100644 --- a/pkg/api/componentreadiness/dataprovider/postgres/provider.go +++ b/pkg/api/componentreadiness/dataprovider/postgres/provider.go @@ -395,6 +395,8 @@ FROM target_tests tt JOIN prow_job_run_tests pjrt ON pjrt.test_id = tt.test_id AND (tt.suite_id = pjrt.suite_id OR (tt.suite_id IS NULL AND pjrt.suite_id IS NULL)) JOIN prow_job_runs pjr ON pjr.id = pjrt.prow_job_run_id + AND pjr.prow_job_release = pjrt.prow_job_run_release + AND pjr.timestamp = pjrt.prow_job_run_timestamp JOIN prow_jobs pj ON pj.id = pjr.prow_job_id JOIN tests t ON t.id = pjrt.test_id WHERE pj.release = ? diff --git a/pkg/api/jobrunscan/reevaluate.go b/pkg/api/jobrunscan/reevaluate.go index 92123ec94..9b610438f 100644 --- a/pkg/api/jobrunscan/reevaluate.go +++ b/pkg/api/jobrunscan/reevaluate.go @@ -542,7 +542,10 @@ func (r *ReEvaluator) updatePostgresLabels(ctx context.Context, buildID string, // subtraction now (idempotent -- a no-op if the row already carries // the label). The label itself is written by the full-array replace // below, so the merged set is used as-is. - if err := infrafailure.SubtractNewInfraFailure(tx, int64(jobRun.ID)); err != nil { //nolint:gosec // G115: prow_job_runs.id is a PostgreSQL serial, always within int64 range + if err := infrafailure.SubtractNewInfraFailure(tx, int64(jobRun.ID), query.ProwJobRunPartitionKeys{ //nolint:gosec // G115: prow_job_runs.id is a PostgreSQL serial, always within int64 range + ProwJobRelease: jobRun.ProwJobRelease, + Timestamp: jobRun.Timestamp, + }); err != nil { return fmt.Errorf("subtracting infra-failure summaries for build %s: %w", buildID, err) } } else if slices.Contains(currentRun.Labels, infrafailure.LabelInfraFailure) { diff --git a/pkg/api/recent_test_failures.go b/pkg/api/recent_test_failures.go index 7b2dea322..d3544e56e 100644 --- a/pkg/api/recent_test_failures.go +++ b/pkg/api/recent_test_failures.go @@ -247,7 +247,9 @@ func fetchOutputs( } if err := dbc.DB.Table("prow_job_run_tests pjrt"). - Joins("JOIN prow_job_runs pjr ON pjr.id = pjrt.prow_job_run_id"). + Joins(`JOIN prow_job_runs pjr ON pjr.id = pjrt.prow_job_run_id + AND pjr.prow_job_release = pjrt.prow_job_run_release + AND pjr.timestamp = pjrt.prow_job_run_timestamp`). Joins("JOIN prow_jobs pj ON pj.id = pjr.prow_job_id"). Joins(`LEFT JOIN prow_job_run_test_outputs pjrto ON pjrto.prow_job_run_test_id = pjrt.id AND pjrto.prow_job_run_test_release = pjrt.prow_job_run_release diff --git a/pkg/dataloader/prowloader/pgwriter/pgwriter.go b/pkg/dataloader/prowloader/pgwriter/pgwriter.go index 1c2962f23..5916244c4 100644 --- a/pkg/dataloader/prowloader/pgwriter/pgwriter.go +++ b/pkg/dataloader/prowloader/pgwriter/pgwriter.go @@ -280,7 +280,6 @@ func insertJobRunIDMap(ctx context.Context, tx pgx.Tx) error { INSERT INTO prow_job_run_id_map (id, prow_job_release, timestamp) SELECT id, prow_job_release, timestamp FROM tmp_prow_job_runs - ON CONFLICT (id) DO NOTHING `); err != nil { return fmt.Errorf("inserting prow_job_run_id_map: %w", err) } diff --git a/pkg/dataloader/prowloader/prow.go b/pkg/dataloader/prowloader/prow.go index 13809c112..048be2cc7 100644 --- a/pkg/dataloader/prowloader/prow.go +++ b/pkg/dataloader/prowloader/prow.go @@ -545,8 +545,8 @@ func (pl *ProwLoader) findNewJobRunIDs(ctx context.Context, candidateIDs []uint) rows, err := conn.Query(ctx, ` SELECT t.id FROM tmp_candidate_ids t - LEFT JOIN prow_job_runs r ON r.id = t.id - WHERE r.id IS NULL + LEFT JOIN prow_job_run_id_map m ON m.id = t.id + WHERE m.id IS NULL `) if err != nil { return nil, fmt.Errorf("querying new job run IDs: %w", err) diff --git a/pkg/db/functions.go b/pkg/db/functions.go index c47615388..cc76d5411 100644 --- a/pkg/db/functions.go +++ b/pkg/db/functions.go @@ -86,6 +86,8 @@ WITH retests AS ( SELECT prow_jobs.id, COUNT(*) as cnt FROM prow_job_runs INNER JOIN prow_job_run_prow_pull_requests on prow_job_run_prow_pull_requests.prow_job_run_id = prow_job_runs.id + AND prow_job_run_prow_pull_requests.prow_job_run_release = prow_job_runs.prow_job_release + AND prow_job_run_prow_pull_requests.prow_job_run_timestamp = prow_job_runs.timestamp INNER JOIN prow_pull_requests on prow_pull_requests.id = prow_job_run_prow_pull_requests.prow_pull_request_id INNER JOIN prow_jobs ON prow_job_runs.prow_job_id = prow_jobs.id WHERE prow_pull_requests.merged_at BETWEEN p_start AND p_endstamp @@ -168,6 +170,8 @@ FROM results SELECT prow_pull_requests.org, prow_pull_requests.repo, prow_jobs.id FROM prow_job_runs INNER JOIN prow_job_run_prow_pull_requests on prow_job_run_prow_pull_requests.prow_job_run_id = prow_job_runs.id + AND prow_job_run_prow_pull_requests.prow_job_run_release = prow_job_runs.prow_job_release + AND prow_job_run_prow_pull_requests.prow_job_run_timestamp = prow_job_runs.timestamp INNER JOIN prow_pull_requests on prow_pull_requests.id = prow_job_run_prow_pull_requests.prow_pull_request_id INNER JOIN prow_jobs ON prow_job_runs.prow_job_id = prow_jobs.id WHERE prow_job_runs.prow_job_release = p_release diff --git a/pkg/db/infrafailure/infrafailure.go b/pkg/db/infrafailure/infrafailure.go index fa6494dac..613d77ddf 100644 --- a/pkg/db/infrafailure/infrafailure.go +++ b/pkg/db/infrafailure/infrafailure.go @@ -13,6 +13,7 @@ package infrafailure import ( "context" + "errors" "fmt" log "github.com/sirupsen/logrus" @@ -81,6 +82,8 @@ const setInfraFailureLabelSQL = ` UPDATE prow_job_runs SET labels = array_append(labels, 'InfraFailure') WHERE id = ? + AND prow_job_release = ? + AND timestamp = ? AND (labels IS NULL OR NOT (labels @> ARRAY['InfraFailure']))` // createDeltasTempTableSQL materializes one job run's test results into a temp @@ -192,10 +195,19 @@ func RecordInfraFailureWithOutcome(ctx context.Context, dbc *gorm.DB, prowJobRun func recordInfraFailureInTx(tx *gorm.DB, prowJobRunID int64) (RecordOutcome, error) { logger := log.WithField("prowJobRunID", prowJobRunID) + partKeys, err := query.LookupProwJobRunPartitionKeys(tx, prowJobRunID) + if err != nil { + if errors.Is(err, gorm.ErrRecordNotFound) { + logger.Debug("prow job run not found in id map; nothing to label") + return OutcomeRunNotFound, nil + } + return OutcomeUnknown, fmt.Errorf("reading partition keys for prow_job_run %d: %w", prowJobRunID, err) + } + // Conditional UPDATE as the first operation: set the label and acquire the // row lock together. RowsAffected == 0 means either the label was already set // (subtraction already done) or the run does not exist in PostgreSQL. - res := tx.Exec(setInfraFailureLabelSQL, prowJobRunID) + res := tx.Exec(setInfraFailureLabelSQL, prowJobRunID, partKeys.ProwJobRelease, partKeys.Timestamp) if res.Error != nil { return OutcomeUnknown, fmt.Errorf("setting InfraFailure label on prow_job_run %d: %w", prowJobRunID, res.Error) } @@ -204,7 +216,9 @@ func recordInfraFailureInTx(tx *gorm.DB, prowJobRunID int64) (RecordOutcome, err // check (cheap, and the row lock the UPDATE would have taken is moot here) // so callers can distinguish an already-labeled run from a missing one. var exists int - check := tx.Raw("SELECT 1 FROM prow_job_runs WHERE id = ? LIMIT 1", prowJobRunID).Scan(&exists) + check := tx.Raw( + "SELECT 1 FROM prow_job_runs WHERE id = ? AND prow_job_release = ? AND timestamp = ? LIMIT 1", + prowJobRunID, partKeys.ProwJobRelease, partKeys.Timestamp).Scan(&exists) if check.Error != nil { return OutcomeUnknown, fmt.Errorf("checking existence of prow_job_run %d: %w", prowJobRunID, check.Error) } @@ -219,7 +233,7 @@ func recordInfraFailureInTx(tx *gorm.DB, prowJobRunID int64) (RecordOutcome, err // The atomic gate passed (the label was newly applied), so remove the run's // contribution from the summary tables in the same transaction. - if err := subtractFromSummaries(tx, prowJobRunID); err != nil { + if err := subtractFromSummaries(tx, prowJobRunID, partKeys); err != nil { return OutcomeUnknown, err } return OutcomeSubtracted, nil @@ -231,17 +245,9 @@ func recordInfraFailureInTx(tx *gorm.DB, prowJobRunID int64) (RecordOutcome, err // own the label and any gating that decides whether the subtraction should run. // All work happens on the supplied transaction so it commits or rolls back // atomically with the caller's other writes. -func subtractFromSummaries(tx *gorm.DB, prowJobRunID int64) error { +func subtractFromSummaries(tx *gorm.DB, prowJobRunID int64, partKeys query.ProwJobRunPartitionKeys) error { logger := log.WithField("prowJobRunID", prowJobRunID) - // Read the run's partition keys (release and timestamp) so the delta scan - // below can prune to the run's single partition instead of scanning every - // partition for the run id. - partKeys, err := query.LookupProwJobRunPartitionKeys(tx, prowJobRunID) - if err != nil { - return fmt.Errorf("reading partition keys for prow_job_run %d: %w", prowJobRunID, err) - } - // Drop any stale temp table before recreating it. ON COMMIT DROP ties the // table's lifetime to the outermost transaction, not to a savepoint, so when // the subtraction runs twice inside the same outer transaction (tx is @@ -291,11 +297,11 @@ func subtractFromSummaries(tx *gorm.DB, prowJobRunID int64) error { // Callers must run this inside the same row-locked transaction that later // replaces the labels array so the containment check and the subtraction stay // consistent with a concurrent RecordInfraFailure. -func SubtractNewInfraFailure(tx *gorm.DB, prowJobRunID int64) error { +func SubtractNewInfraFailure(tx *gorm.DB, prowJobRunID int64, partKeys query.ProwJobRunPartitionKeys) error { var found int res := tx.Raw( - "SELECT 1 FROM prow_job_runs WHERE id = ? AND labels @> ARRAY[?] LIMIT 1", - prowJobRunID, LabelInfraFailure).Scan(&found) + "SELECT 1 FROM prow_job_runs WHERE id = ? AND prow_job_release = ? AND timestamp = ? AND labels @> ARRAY[?] LIMIT 1", + prowJobRunID, partKeys.ProwJobRelease, partKeys.Timestamp, LabelInfraFailure).Scan(&found) if res.Error != nil { return fmt.Errorf("checking InfraFailure label on prow_job_run %d: %w", prowJobRunID, res.Error) } @@ -303,5 +309,5 @@ func SubtractNewInfraFailure(tx *gorm.DB, prowJobRunID int64) error { log.WithField("prowJobRunID", prowJobRunID).Debug("prow job run already labeled InfraFailure; summary subtraction already applied, skipping") return nil } - return subtractFromSummaries(tx, prowJobRunID) + return subtractFromSummaries(tx, prowJobRunID, partKeys) } diff --git a/pkg/db/query/job_queries.go b/pkg/db/query/job_queries.go index 424aa5544..21ac8b929 100644 --- a/pkg/db/query/job_queries.go +++ b/pkg/db/query/job_queries.go @@ -15,19 +15,19 @@ import ( // ProwJobRunPartitionKeys holds the partition key columns for prow_job_runs. // Used for two-step lookups: fetch these lightweight keys first, then load the -// full row with partition pruning. This will be replaced by a mapping table in -// a future iteration. +// full row with partition pruning. type ProwJobRunPartitionKeys struct { ProwJobRelease string `gorm:"column:prow_job_release"` Timestamp time.Time `gorm:"column:timestamp"` } // LookupProwJobRunPartitionKeys fetches the partition keys for a prow_job_run -// by ID. This is intended as the first step of a two-step lookup pattern where -// the caller then uses these keys to load the full row with partition pruning. +// by ID from prow_job_run_id_map. This is the first step of a two-step lookup +// pattern where the caller then uses these keys to load the full row with +// partition pruning. func LookupProwJobRunPartitionKeys(gormDB *gorm.DB, jobRunID int64) (ProwJobRunPartitionKeys, error) { var keys ProwJobRunPartitionKeys - err := gormDB.Table("prow_job_runs"). + err := gormDB.Model(&models.ProwJobRunIDMap{}). Select("prow_job_release, timestamp"). Where("id = ?", jobRunID). Take(&keys).Error diff --git a/pkg/db/query/pull_request_queries.go b/pkg/db/query/pull_request_queries.go index 50ee698bf..fb77fcd05 100644 --- a/pkg/db/query/pull_request_queries.go +++ b/pkg/db/query/pull_request_queries.go @@ -26,7 +26,7 @@ func PullRequestReport(dbc *db.DB, filterOpts *filter.FilterOptions, release str Joins("LEFT JOIN (?) nightly ON nightly.url = prow_pull_requests.link", dbc.DB.Table("(?) as nightly", firstPayloadsByStreamAndArch).Where("nightly.stream = 'nightly' AND nightly.architecture = 'amd64'")). Joins("INNER JOIN prow_job_run_prow_pull_requests ON prow_job_run_prow_pull_requests.prow_pull_request_id = prow_pull_requests.id"). - Joins("INNER JOIN prow_job_runs on prow_job_run_prow_pull_requests.prow_job_run_id = prow_job_runs.id"). + Joins("INNER JOIN prow_job_runs on prow_job_run_prow_pull_requests.prow_job_run_id = prow_job_runs.id AND prow_job_run_prow_pull_requests.prow_job_run_release = prow_job_runs.prow_job_release AND prow_job_run_prow_pull_requests.prow_job_run_timestamp = prow_job_runs.timestamp"). Joins("INNER JOIN prow_jobs on prow_job_runs.prow_job_id = prow_jobs.id"). Where("prow_jobs.release = ?", release). Where("prow_job_runs.prow_job_release = ?", release). @@ -49,7 +49,7 @@ func PullRequestReport(dbc *db.DB, filterOpts *filter.FilterOptions, release str func PullRequestAveragePremergeFailures(dbc *db.DB, release string, start, end *time.Time) *gorm.DB { premergeFailures := dbc.DB.Table("prow_job_runs"). Select("prow_jobs.id as prow_job_id, prow_jobs.name as prow_job_name, prow_pull_requests.org, prow_pull_requests.repo, prow_pull_requests.link, COUNT(*) as total_runs"). - Joins("INNER JOIN prow_job_run_prow_pull_requests on prow_job_run_prow_pull_requests.prow_job_run_id = prow_job_runs.id AND prow_job_run_prow_pull_requests.prow_job_run_release = prow_job_runs.prow_job_release"). + Joins("INNER JOIN prow_job_run_prow_pull_requests on prow_job_run_prow_pull_requests.prow_job_run_id = prow_job_runs.id AND prow_job_run_prow_pull_requests.prow_job_run_release = prow_job_runs.prow_job_release AND prow_job_run_prow_pull_requests.prow_job_run_timestamp = prow_job_runs.timestamp"). Joins("INNER JOIN prow_pull_requests on prow_pull_requests.id = prow_job_run_prow_pull_requests.prow_pull_request_id"). Joins("INNER JOIN prow_jobs ON prow_job_runs.prow_job_id = prow_jobs.id"). Where("prow_job_runs.prow_job_release = ?", release). diff --git a/pkg/db/query/repository_queries.go b/pkg/db/query/repository_queries.go index 36990b5bb..fa9f4faf9 100644 --- a/pkg/db/query/repository_queries.go +++ b/pkg/db/query/repository_queries.go @@ -21,7 +21,7 @@ func RepositoryReport(dbc *db.DB, filterOpts *filter.FilterOptions, release stri repos := dbc.DB.Table("prow_pull_requests"). Joins("INNER JOIN prow_job_run_prow_pull_requests ON prow_job_run_prow_pull_requests.prow_pull_request_id = prow_pull_requests.id"). - Joins("INNER JOIN prow_job_runs on prow_job_run_prow_pull_requests.prow_job_run_id = prow_job_runs.id"). + Joins("INNER JOIN prow_job_runs on prow_job_run_prow_pull_requests.prow_job_run_id = prow_job_runs.id AND prow_job_run_prow_pull_requests.prow_job_run_release = prow_job_runs.prow_job_release AND prow_job_run_prow_pull_requests.prow_job_run_timestamp = prow_job_runs.timestamp"). Joins("INNER JOIN prow_jobs on prow_job_runs.prow_job_id = prow_jobs.id"). Joins("LEFT JOIN (?) revert_count ON revert_count.org = prow_pull_requests.org AND revert_count.repo = prow_pull_requests.repo", revertCount). Joins("LEFT JOIN (?) premerge_failures ON premerge_failures.prow_job_ID = prow_jobs.id", averageByJob). diff --git a/pkg/sippyserver/server.go b/pkg/sippyserver/server.go index 7a481b9fc..c89ce7711 100644 --- a/pkg/sippyserver/server.go +++ b/pkg/sippyserver/server.go @@ -1036,7 +1036,7 @@ func (s *Server) jsonBackendDisruptionByRun(w http.ResponseWriter, req *http.Req var minTime, maxTime time.Time if s.db != nil { - row := s.db.DB.Raw("SELECT MIN(timestamp), MAX(timestamp) FROM prow_job_runs WHERE id IN ?", jobRunNames).Row() + row := s.db.DB.Raw("SELECT MIN(timestamp), MAX(timestamp) FROM prow_job_run_id_map WHERE id IN ?", jobRunNames).Row() if err := row.Scan(&minTime, &maxTime); err != nil { log.WithError(err).Warn("could not look up job run timestamps from postgres, falling back to no time bound") } diff --git a/test/integration/job_run_id_map_test.go b/test/integration/job_run_id_map_test.go new file mode 100644 index 000000000..ed67c2e8e --- /dev/null +++ b/test/integration/job_run_id_map_test.go @@ -0,0 +1,29 @@ +package integration + +import ( + "testing" + "time" + + "github.com/stretchr/testify/assert" + "github.com/stretchr/testify/require" + "gorm.io/gorm" + + v1 "github.com/openshift/sippy/pkg/apis/sippyprocessing/v1" + "github.com/openshift/sippy/pkg/db/query" + intutil "github.com/openshift/sippy/test/integration/util" +) + +func TestLookupProwJobRunPartitionKeys(t *testing.T) { + dbc := intutil.NewTestDB(t, pgContainer) + job := intutil.CreateProwJob(t, dbc, "periodic-e2e-aws", "4.18", nil) + ts := time.Date(2026, 7, 15, 10, 0, 0, 0, time.UTC) + run := intutil.CreateProwJobRun(t, dbc, job.ID, "4.18", ts, true, v1.JobSucceeded) + + keys, err := query.LookupProwJobRunPartitionKeys(dbc.DB, int64(run.ID)) + require.NoError(t, err) + assert.Equal(t, "4.18", keys.ProwJobRelease) + assert.True(t, ts.Equal(keys.Timestamp)) + + _, err = query.LookupProwJobRunPartitionKeys(dbc.DB, 999999) + require.ErrorIs(t, err, gorm.ErrRecordNotFound) +}