diff --git a/cmd/sippy/backfill.go b/cmd/sippy/backfill.go new file mode 100644 index 0000000000..8178ed9c8d --- /dev/null +++ b/cmd/sippy/backfill.go @@ -0,0 +1,102 @@ +package main + +import ( + "fmt" + "time" + + "cloud.google.com/go/civil" + log "github.com/sirupsen/logrus" + "github.com/spf13/cobra" + "github.com/spf13/pflag" + "k8s.io/apimachinery/pkg/util/sets" + + "github.com/openshift/sippy/pkg/flags" + "github.com/openshift/sippy/pkg/sippyserver" +) + +type BackfillFlags struct { + DBFlags *flags.PostgresFlags + Table string + StartDate string + EndDate string +} + +func NewBackfillFlags() *BackfillFlags { + return &BackfillFlags{ + DBFlags: flags.NewPostgresDatabaseFlags(), + } +} + +func (f *BackfillFlags) BindFlags(fs *pflag.FlagSet) { + f.DBFlags.BindFlags(fs) + fs.StringVar(&f.Table, "table", "", "Table to backfill (daily-summaries, daily-totals, cumulative-summaries)") + fs.StringVar(&f.StartDate, "start-date", "", "Start date (YYYY-MM-DD)") + fs.StringVar(&f.EndDate, "end-date", "", "End date (YYYY-MM-DD)") +} + +func NewBackfillCommand() *cobra.Command { + f := NewBackfillFlags() + + cmd := &cobra.Command{ + Use: "backfill", + Short: "Backfill a specific summary table for a date range", + RunE: func(cmd *cobra.Command, args []string) error { + validTables := sets.New[string]("daily-summaries", "daily-totals", "cumulative-summaries") + if f.Table == "" { + return fmt.Errorf("--table is required") + } + if !validTables.Has(f.Table) { + return fmt.Errorf("invalid --table %q: must be one of %v", f.Table, sets.List(validTables)) + } + if f.StartDate == "" { + return fmt.Errorf("--start-date is required") + } + + startDate, err := civil.ParseDate(f.StartDate) + if err != nil { + return fmt.Errorf("invalid --start-date %q: %w", f.StartDate, err) + } + + endDate := civil.DateOf(time.Now().UTC()) + if f.EndDate != "" { + endDate, err = civil.ParseDate(f.EndDate) + if err != nil { + return fmt.Errorf("invalid --end-date %q: %w", f.EndDate, err) + } + } + + if startDate.After(endDate) { + return fmt.Errorf("--start-date (%s) is after --end-date (%s)", startDate, endDate) + } + + dbc, err := f.DBFlags.GetDBClient() + if err != nil { + return fmt.Errorf("getting db client: %w", err) + } + + log.WithFields(log.Fields{ + "table": f.Table, + "start": startDate, + "end": endDate, + }).Info("starting backfill") + + var releases []string + if err := dbc.DB.Table("release_definitions"). + Order("major DESC, minor DESC"). + Pluck("release", &releases).Error; err != nil { + return fmt.Errorf("querying releases: %w", err) + } + startTime := time.Date(startDate.Year, startDate.Month, startDate.Day, 0, 0, 0, 0, time.UTC) + endTime := time.Date(endDate.Year, endDate.Month, endDate.Day, 0, 0, 0, 0, time.UTC).AddDate(0, 0, 1) + if _, err := dbc.EnsurePartitions(releases, startTime, endTime, false); err != nil { + return fmt.Errorf("ensuring partitions: %w", err) + } + + return sippyserver.BackfillData(dbc, f.Table, startDate, endDate) + }, + } + + f.BindFlags(cmd.Flags()) + + return cmd +} diff --git a/cmd/sippy/load.go b/cmd/sippy/load.go index 77be99579c..c45c408734 100644 --- a/cmd/sippy/load.go +++ b/cmd/sippy/load.go @@ -41,7 +41,6 @@ import ( "github.com/openshift/sippy/pkg/dataloader/releaseloader" "github.com/openshift/sippy/pkg/dataloader/testownershiploader" "github.com/openshift/sippy/pkg/db" - "github.com/openshift/sippy/pkg/db/dailysummary" "github.com/openshift/sippy/pkg/flags" "github.com/openshift/sippy/pkg/github/commenter" ) @@ -371,7 +370,10 @@ func NewLoadCommand() *cobra.Command { log.WithField("elapsed", elapsed).Info("database load complete") if refreshMatviews && !f.SkipMatviewRefresh { - sippyserver.RefreshData(dbc, cacheClient, false, dailysummary.Options{}) + if err := sippyserver.RefreshData(dbc, cacheClient, sippyserver.RefreshOptions{}); err != nil { + log.WithError(err).Error("refresh failed") + allErrs = append(allErrs, err) + } } elapsed = time.Since(start) diff --git a/cmd/sippy/main.go b/cmd/sippy/main.go index fd9ab18d35..e0266fb1c9 100644 --- a/cmd/sippy/main.go +++ b/cmd/sippy/main.go @@ -39,6 +39,7 @@ func main() { NewLoadCommand(), NewSnapshotCommand(), NewRefreshCommand(), + NewBackfillCommand(), NewComponentReadinessCommand(), NewAutomateJiraCommand(), NewVariantsCommand(), diff --git a/cmd/sippy/refresh.go b/cmd/sippy/refresh.go index e9cdaedeb7..ed0d9d3a93 100644 --- a/cmd/sippy/refresh.go +++ b/cmd/sippy/refresh.go @@ -1,25 +1,18 @@ package main import ( - "fmt" - "time" - "github.com/sirupsen/logrus" "github.com/spf13/cobra" "github.com/spf13/pflag" - "github.com/openshift/sippy/pkg/db/dailysummary" "github.com/openshift/sippy/pkg/flags" "github.com/openshift/sippy/pkg/sippyserver" ) type RefreshFlags struct { - DBFlags *flags.PostgresFlags - CacheFlags *flags.CacheFlags - RefreshOnlyIfEmpty bool - RebuildDailySummaries bool - DailySummariesStartDate string - DailySummariesEndDate string + DBFlags *flags.PostgresFlags + CacheFlags *flags.CacheFlags + RefreshOnlyIfEmpty bool } func NewRefreshFlags() *RefreshFlags { @@ -33,32 +26,6 @@ func (f *RefreshFlags) BindFlags(fs *pflag.FlagSet) { f.DBFlags.BindFlags(fs) f.CacheFlags.BindFlags(fs) fs.BoolVar(&f.RefreshOnlyIfEmpty, "refresh-only-if-empty", f.RefreshOnlyIfEmpty, "only refresh matviews if they're empty") - fs.BoolVar(&f.RebuildDailySummaries, "rebuild-daily-summaries", false, "Truncate and rebuild the daily summaries table for the default lookback period (use --daily-summaries-start for older data)") - fs.StringVar(&f.DailySummariesStartDate, "daily-summaries-start", "", "Override start date for daily summaries (YYYY-MM-DD)") - fs.StringVar(&f.DailySummariesEndDate, "daily-summaries-end", "", "Override end date for daily summaries (YYYY-MM-DD)") -} - -func (f *RefreshFlags) dailySummaryOptions() (dailysummary.Options, error) { - opts := dailysummary.Options{Rebuild: f.RebuildDailySummaries} - if f.DailySummariesStartDate != "" { - t, err := time.Parse("2006-01-02", f.DailySummariesStartDate) - if err != nil { - return opts, fmt.Errorf("invalid --daily-summaries-start %q: %w", f.DailySummariesStartDate, err) - } - opts.StartOverride = &t - } - if f.DailySummariesEndDate != "" { - t, err := time.Parse("2006-01-02", f.DailySummariesEndDate) - if err != nil { - return opts, fmt.Errorf("invalid --daily-summaries-end %q: %w", f.DailySummariesEndDate, err) - } - opts.EndOverride = &t - } - if opts.StartOverride != nil && opts.EndOverride != nil && opts.StartOverride.After(*opts.EndOverride) { - return opts, fmt.Errorf("--daily-summaries-start (%s) is after --daily-summaries-end (%s)", - f.DailySummariesStartDate, f.DailySummariesEndDate) - } - return opts, nil } func NewRefreshCommand() *cobra.Command { @@ -79,12 +46,9 @@ func NewRefreshCommand() *cobra.Command { } else if cacheClient == nil { logrus.Warn("no cache provided; refresh will not update cached timestamps, so cached data may not be properly invalidated") } - dailySummaryOpts, err := f.dailySummaryOptions() - if err != nil { - return err - } - sippyserver.RefreshData(dbc, cacheClient, f.RefreshOnlyIfEmpty, dailySummaryOpts) - return nil + return sippyserver.RefreshData(dbc, cacheClient, sippyserver.RefreshOptions{ + RefreshOnlyIfEmpty: f.RefreshOnlyIfEmpty, + }) }, } diff --git a/cmd/sippy/seed_data.go b/cmd/sippy/seed_data.go index b8736bc5cf..4b4b97b306 100644 --- a/cmd/sippy/seed_data.go +++ b/cmd/sippy/seed_data.go @@ -9,6 +9,7 @@ import ( "strings" "time" + "cloud.google.com/go/civil" "github.com/lib/pq" "github.com/pkg/errors" log "github.com/sirupsen/logrus" @@ -23,7 +24,6 @@ import ( "github.com/openshift/sippy/pkg/apis/api/componentreport/reqopts" v1 "github.com/openshift/sippy/pkg/apis/sippyprocessing/v1" "github.com/openshift/sippy/pkg/db" - "github.com/openshift/sippy/pkg/db/dailysummary" "github.com/openshift/sippy/pkg/db/models" "github.com/openshift/sippy/pkg/db/models/jobrunscan" "github.com/openshift/sippy/pkg/flags" @@ -439,8 +439,17 @@ func seedSyntheticData(dbc *db.DB) error { } log.Info("Refreshing materialized views...") - seedStart := time.Now().Add(-190 * 24 * time.Hour) - sippyserver.RefreshData(dbc, nil, false, dailysummary.Options{StartOverride: &seedStart}) + seedToday := civil.DateOf(time.Now().UTC()) + seedStart := seedToday.AddDays(-190) + seedEnd := seedToday + for _, table := range []string{"daily-summaries", "daily-totals", "cumulative-summaries"} { + if err := sippyserver.BackfillData(dbc, table, seedStart, seedEnd); err != nil { + return fmt.Errorf("failed to backfill %s: %w", table, err) + } + } + if err := sippyserver.RefreshData(dbc, nil, sippyserver.RefreshOptions{}); err != nil { + return fmt.Errorf("failed to refresh data: %w", err) + } log.Info("Syncing regressions...") if err := syncRegressions(dbc); err != nil { diff --git a/pkg/db/cumulativesummary/cumulative_summary.go b/pkg/db/cumulativesummary/cumulative_summary.go new file mode 100644 index 0000000000..a5da5a8e5a --- /dev/null +++ b/pkg/db/cumulativesummary/cumulative_summary.go @@ -0,0 +1,161 @@ +package cumulativesummary + +import ( + "fmt" + "time" + + "cloud.google.com/go/civil" + log "github.com/sirupsen/logrus" + + "github.com/openshift/sippy/pkg/db" +) + +type summaryStore interface { + MaxCumulativeSummaryDate() (*civil.Date, error) + MaxDailySummaryDate() (*civil.Date, error) + Releases() ([]string, error) + UpdateDateForRelease(date civil.Date, release string) error +} + +// Refresh computes cumulative summaries from test_daily_totals. +// earliestChanged is the earliest date that test_daily_totals was +// updated; cumulative summaries from that date forward will be recomputed. +// Returns the earliest date that was updated so downstream consumers +// (variant cumulative summaries) know where to start. +func Refresh(dbc *db.DB, earliestChanged civil.Date) (civil.Date, error) { + return doRefresh(&pgStore{dbc: dbc}, earliestChanged) +} + +// Backfill processes an explicit date range without automatic date detection. +func Backfill(dbc *db.DB, startDate, endDate civil.Date) error { + store := &pgStore{dbc: dbc} + _, err := refreshDateRange(store, startDate, endDate) + return err +} + +func doRefresh(store summaryStore, earliestChanged civil.Date) (civil.Date, error) { + return doRefreshWithToday(store, earliestChanged, civil.DateOf(time.Now().UTC())) +} + +func doRefreshWithToday(store summaryStore, earliestChanged, today civil.Date) (civil.Date, error) { + log.Info("refreshing cumulative summaries") + + maxCumulativeSummaryDate, err := store.MaxCumulativeSummaryDate() + if err != nil { + return civil.Date{}, fmt.Errorf("checking max cumulative summary date: %w", err) + } + + maxDailySummaryDate, err := store.MaxDailySummaryDate() + if err != nil { + return civil.Date{}, fmt.Errorf("checking max daily summary date: %w", err) + } + if maxDailySummaryDate == nil { + log.Info("no daily summaries found, nothing to update") + return earliestChanged, nil + } + endDate := *maxDailySummaryDate + + startDate := resolveStartDate(earliestChanged, maxCumulativeSummaryDate, today) + + return refreshDateRange(store, startDate, endDate) +} + +func refreshDateRange(store summaryStore, startDate, endDate civil.Date) (civil.Date, error) { + loadStart := time.Now() + days := endDate.DaysSince(startDate) + 1 + + releases, err := store.Releases() + if err != nil { + return civil.Date{}, fmt.Errorf("querying releases: %w", err) + } + + log.WithFields(log.Fields{ + "start": startDate, + "end": endDate, + "days": days, + "releases": len(releases), + }).Info("updating cumulative summaries") + + for date := startDate; !date.After(endDate); date = date.AddDays(1) { + updateStart := time.Now() + for _, release := range releases { + if err := store.UpdateDateForRelease(date, release); err != nil { + return civil.Date{}, fmt.Errorf("updating cumulative summaries for %s release %s: %w", date, release, err) + } + } + log.WithFields(log.Fields{ + "date": date, + "elapsed": time.Since(updateStart), + }).Info("updated cumulative summaries for date") + } + + log.WithField("elapsed", time.Since(loadStart)).Info("cumulative summary refresh complete") + return startDate, nil +} + +type pgStore struct { + dbc *db.DB +} + +func (s *pgStore) MaxCumulativeSummaryDate() (*civil.Date, error) { + var d *civil.Date + err := s.dbc.DB.Table("test_cumulative_summaries"). + Select("MAX(date)").Row().Scan(&d) + if err != nil { + return nil, err + } + + return d, nil +} + +func (s *pgStore) MaxDailySummaryDate() (*civil.Date, error) { + var d *civil.Date + err := s.dbc.DB.Table("test_daily_totals"). + Select("MAX(date)").Row().Scan(&d) + if err != nil { + return nil, err + } + + return d, nil +} + +func (s *pgStore) Releases() ([]string, error) { + var releases []string + err := s.dbc.DB.Table("release_definitions"). + Order("major DESC, minor DESC"). + Pluck("release", &releases).Error + return releases, err +} + +func (s *pgStore) UpdateDateForRelease(date civil.Date, release string) error { + return s.dbc.DB.Exec(` + INSERT INTO test_cumulative_summaries (date, test_id, prow_job_id, suite_id, release, + prefix_sum_successes, prefix_sum_failures, prefix_sum_flakes, prefix_sum_runs) + SELECT + ?, + COALESCE(prev.test_id, tds.test_id), + COALESCE(prev.prow_job_id, tds.prow_job_id), + COALESCE(prev.suite_id, tds.suite_id), + COALESCE(prev.release, tds.release), + COALESCE(prev.prefix_sum_successes, 0) + COALESCE(tds.successes, 0), + COALESCE(prev.prefix_sum_failures, 0) + COALESCE(tds.failures, 0), + COALESCE(prev.prefix_sum_flakes, 0) + COALESCE(tds.flakes, 0), + COALESCE(prev.prefix_sum_runs, 0) + COALESCE(tds.runs, 0) + FROM (SELECT * FROM test_cumulative_summaries WHERE date = ?::date - 1 AND release = ?) prev + FULL OUTER JOIN (SELECT * FROM test_daily_totals WHERE date = ? AND release = ?) tds + ON prev.test_id = tds.test_id + AND prev.prow_job_id = tds.prow_job_id + AND prev.suite_id = tds.suite_id + ON CONFLICT (date, release, test_id, prow_job_id, suite_id) + DO UPDATE SET + prefix_sum_successes = EXCLUDED.prefix_sum_successes, + prefix_sum_failures = EXCLUDED.prefix_sum_failures, + prefix_sum_flakes = EXCLUDED.prefix_sum_flakes, + prefix_sum_runs = EXCLUDED.prefix_sum_runs + WHERE (test_cumulative_summaries.prefix_sum_successes, test_cumulative_summaries.prefix_sum_failures, + test_cumulative_summaries.prefix_sum_flakes, test_cumulative_summaries.prefix_sum_runs) + IS DISTINCT FROM + (EXCLUDED.prefix_sum_successes, EXCLUDED.prefix_sum_failures, + EXCLUDED.prefix_sum_flakes, EXCLUDED.prefix_sum_runs) + `, date, date, release, date, release).Error +} diff --git a/pkg/db/cumulativesummary/cumulative_summary_test.go b/pkg/db/cumulativesummary/cumulative_summary_test.go new file mode 100644 index 0000000000..227726785d --- /dev/null +++ b/pkg/db/cumulativesummary/cumulative_summary_test.go @@ -0,0 +1,150 @@ +package cumulativesummary + +import ( + "fmt" + "testing" + + "cloud.google.com/go/civil" + "github.com/stretchr/testify/assert" + "github.com/stretchr/testify/require" +) + +type fakeStore struct { + maxCumulativeSummary *civil.Date + maxCumulativeSummaryErr error + maxDailySummaryDate *civil.Date + maxDailySummaryDateErr error + updatedDates []civil.Date + updateDateErr error +} + +func (f *fakeStore) MaxCumulativeSummaryDate() (*civil.Date, error) { + return f.maxCumulativeSummary, f.maxCumulativeSummaryErr +} + +func (f *fakeStore) MaxDailySummaryDate() (*civil.Date, error) { + return f.maxDailySummaryDate, f.maxDailySummaryDateErr +} + +func (f *fakeStore) Releases() ([]string, error) { + return []string{"4.22", "5.0"}, nil +} + +func (f *fakeStore) UpdateDateForRelease(date civil.Date, _ string) error { + f.updatedDates = append(f.updatedDates, date) + return f.updateDateErr +} + +var today = civil.Date{Year: 2026, Month: 7, Day: 9} + +const numReleases = 2 + +func TestRefresh_NoCumulativeDataUsesEarliestChanged(t *testing.T) { + maxDailySummaryDate := today + earliestChanged := today.AddDays(-1) + store := &fakeStore{maxDailySummaryDate: &maxDailySummaryDate} + + _, err := doRefreshWithToday(store, earliestChanged, today) + + require.NoError(t, err) + require.NotEmpty(t, store.updatedDates) + assert.Equal(t, earliestChanged, store.updatedDates[0]) +} + +func TestRefresh_FillsGapFromMaxCumulative(t *testing.T) { + maxCum := today.AddDays(-4) + maxDailySummaryDate := today.AddDays(-2) + earliestChanged := today.AddDays(-2) + store := &fakeStore{maxCumulativeSummary: &maxCum, maxDailySummaryDate: &maxDailySummaryDate} + + _, err := doRefreshWithToday(store, earliestChanged, today) + + require.NoError(t, err) + assert.Len(t, store.updatedDates, 2*numReleases) + assert.Equal(t, today.AddDays(-3), store.updatedDates[0]) +} + +func TestRefresh_StartsFromEarliestChanged(t *testing.T) { + maxDailySummaryDate := today.AddDays(-2) + earliestChanged := today.AddDays(-3) + store := &fakeStore{maxDailySummaryDate: &maxDailySummaryDate} + + _, err := doRefreshWithToday(store, earliestChanged, today) + + require.NoError(t, err) + assert.Len(t, store.updatedDates, 2*numReleases) + assert.Equal(t, today.AddDays(-3), store.updatedDates[0]) +} + +func TestRefresh_FillsSmallGap(t *testing.T) { + maxCum := today.AddDays(-6) + maxDailySummaryDate := today.AddDays(-2) + earliestChanged := today.AddDays(-3) + store := &fakeStore{maxCumulativeSummary: &maxCum, maxDailySummaryDate: &maxDailySummaryDate} + + _, err := doRefreshWithToday(store, earliestChanged, today) + + require.NoError(t, err) + assert.Equal(t, today.AddDays(-5), store.updatedDates[0]) + assert.Equal(t, today.AddDays(-2), store.updatedDates[len(store.updatedDates)-1]) +} + +func TestRefresh_CapsGapAt14Days(t *testing.T) { + maxCum := today.AddDays(-30) + maxDailySummaryDate := today + earliestChanged := today.AddDays(-1) + store := &fakeStore{maxCumulativeSummary: &maxCum, maxDailySummaryDate: &maxDailySummaryDate} + + _, err := doRefreshWithToday(store, earliestChanged, today) + + require.NoError(t, err) + assert.Equal(t, today.AddDays(-14), store.updatedDates[0]) +} + +func TestRefresh_IncrementalFromMaxPlusOne(t *testing.T) { + maxCum := today.AddDays(-3) + maxDailySummaryDate := today.AddDays(-2) + earliestChanged := today.AddDays(-3) + store := &fakeStore{maxCumulativeSummary: &maxCum, maxDailySummaryDate: &maxDailySummaryDate} + + _, err := doRefreshWithToday(store, earliestChanged, today) + + require.NoError(t, err) + assert.Len(t, store.updatedDates, 2*numReleases) + assert.Equal(t, today.AddDays(-3), store.updatedDates[0]) +} + +func TestRefresh_AlreadyUpToDate(t *testing.T) { + maxCum := today.AddDays(-2) + maxDailySummaryDate := today.AddDays(-2) + earliestChanged := today.AddDays(-2) + store := &fakeStore{maxCumulativeSummary: &maxCum, maxDailySummaryDate: &maxDailySummaryDate} + + _, err := doRefreshWithToday(store, earliestChanged, today) + + require.NoError(t, err) + assert.Len(t, store.updatedDates, 1*numReleases) + assert.Equal(t, today.AddDays(-2), store.updatedDates[0]) +} + +func TestRefresh_MaxCumulativeSummaryDateError(t *testing.T) { + earliestChanged := today.AddDays(-3) + store := &fakeStore{maxCumulativeSummaryErr: fmt.Errorf("connection refused")} + + _, err := doRefreshWithToday(store, earliestChanged, today) + + require.Error(t, err) + assert.Contains(t, err.Error(), "max cumulative summary date") +} + +func TestRefresh_UpdateDateError(t *testing.T) { + maxCum := today.AddDays(-4) + maxDailySummaryDate := today.AddDays(-2) + earliestChanged := today.AddDays(-3) + store := &fakeStore{maxCumulativeSummary: &maxCum, maxDailySummaryDate: &maxDailySummaryDate, updateDateErr: fmt.Errorf("disk full")} + + _, err := doRefreshWithToday(store, earliestChanged, today) + + require.Error(t, err) + assert.Contains(t, err.Error(), "updating cumulative summaries") +} diff --git a/pkg/db/cumulativesummary/date_range.go b/pkg/db/cumulativesummary/date_range.go new file mode 100644 index 0000000000..1311b7c280 --- /dev/null +++ b/pkg/db/cumulativesummary/date_range.go @@ -0,0 +1,33 @@ +package cumulativesummary + +import ( + "cloud.google.com/go/civil" + log "github.com/sirupsen/logrus" +) + +const maxAutoFillDays = 14 + +// resolveStartDate determines the start date for an incremental refresh. +// It picks the earliest of earliestChanged and maxExistingDate+1, then +// clamps to today-maxAutoFillDays to prevent unbounded backfills during +// normal refresh cycles. +func resolveStartDate(earliestChanged civil.Date, maxExistingDate *civil.Date, today civil.Date) civil.Date { + startDate := earliestChanged + if maxExistingDate != nil { + nextDate := maxExistingDate.AddDays(1) + if nextDate.Before(startDate) { + startDate = nextDate + } + } + + floor := today.AddDays(-maxAutoFillDays) + if startDate.Before(floor) { + log.WithFields(log.Fields{ + "desiredStart": startDate, + "clampedStart": floor, + }).Warn("cumulative summaries need data older than the auto-fill limit, use backfill command to fill") + startDate = floor + } + + return startDate +} diff --git a/pkg/db/cumulativesummary/date_range_test.go b/pkg/db/cumulativesummary/date_range_test.go new file mode 100644 index 0000000000..3ea2da8adc --- /dev/null +++ b/pkg/db/cumulativesummary/date_range_test.go @@ -0,0 +1,61 @@ +package cumulativesummary + +import ( + "testing" + + "cloud.google.com/go/civil" + "github.com/stretchr/testify/assert" +) + +var testToday = civil.Date{Year: 2026, Month: 7, Day: 9} + +func TestResolveStartDate_UsesEarliestChanged(t *testing.T) { + earliestChanged := testToday.AddDays(-3) + + result := resolveStartDate(earliestChanged, nil, testToday) + + assert.Equal(t, earliestChanged, result) +} + +func TestResolveStartDate_UsesMaxPlusOneWhenEarlier(t *testing.T) { + earliestChanged := testToday.AddDays(-1) + maxExisting := testToday.AddDays(-5) + + result := resolveStartDate(earliestChanged, &maxExisting, testToday) + + assert.Equal(t, maxExisting.AddDays(1), result) +} + +func TestResolveStartDate_UsesEarliestChangedWhenEarlierThanMax(t *testing.T) { + earliestChanged := testToday.AddDays(-5) + maxExisting := testToday.AddDays(-2) + + result := resolveStartDate(earliestChanged, &maxExisting, testToday) + + assert.Equal(t, earliestChanged, result) +} + +func TestResolveStartDate_ClampsToFloor(t *testing.T) { + earliestChanged := testToday.AddDays(-30) + + result := resolveStartDate(earliestChanged, nil, testToday) + + assert.Equal(t, testToday.AddDays(-maxAutoFillDays), result) +} + +func TestResolveStartDate_ClampsMaxPlusOneToFloor(t *testing.T) { + earliestChanged := testToday.AddDays(-1) + maxExisting := testToday.AddDays(-30) + + result := resolveStartDate(earliestChanged, &maxExisting, testToday) + + assert.Equal(t, testToday.AddDays(-maxAutoFillDays), result) +} + +func TestResolveStartDate_DoesNotClampWhenWithinFloor(t *testing.T) { + earliestChanged := testToday.AddDays(-7) + + result := resolveStartDate(earliestChanged, nil, testToday) + + assert.Equal(t, earliestChanged, result) +} diff --git a/pkg/db/dailysummary/dailysummary.go b/pkg/db/dailysummary/dailysummary.go index bec180f19c..351bdecd99 100644 --- a/pkg/db/dailysummary/dailysummary.go +++ b/pkg/db/dailysummary/dailysummary.go @@ -7,6 +7,7 @@ import ( "sync" "time" + "cloud.google.com/go/civil" log "github.com/sirupsen/logrus" "github.com/openshift/sippy/pkg/db" @@ -19,14 +20,9 @@ const ( var valueColumns = []string{"successes", "failures", "flakes", "runs"} -var ( - insertSQL = buildInsertSQL() - onConflictClause = buildOnConflictClause() -) - -func buildInsertSQL() string { +func buildInsertSQL(tableName, dateColumn string) string { return fmt.Sprintf(` - INSERT INTO test_daily_summaries (test_id, prow_job_id, suite_id, release, summary_date, %s) + INSERT INTO %s (test_id, prow_job_id, suite_id, release, %s, %s) SELECT pjrt.test_id, pjrt.prow_job_id, @@ -42,89 +38,135 @@ func buildInsertSQL() string { AND pjrt.prow_job_run_timestamp < (?::date + INTERVAL '1 day') AND pjrt.prow_job_run_release = ? GROUP BY pjrt.test_id, pjrt.prow_job_id, COALESCE(pjrt.suite_id, 0), pjrt.prow_job_run_release, date(pjrt.prow_job_run_timestamp)`, - strings.Join(valueColumns, ", ")) + tableName, dateColumn, strings.Join(valueColumns, ", ")) } -func buildOnConflictClause() string { +func buildOnConflictClause(tableName, dateColumn string) string { var setClauses, oldCols, newCols []string for _, col := range valueColumns { setClauses = append(setClauses, fmt.Sprintf("%s = EXCLUDED.%s", col, col)) - oldCols = append(oldCols, "test_daily_summaries."+col) + oldCols = append(oldCols, tableName+"."+col) newCols = append(newCols, "EXCLUDED."+col) } return fmt.Sprintf(` - ON CONFLICT (test_id, prow_job_id, suite_id, release, summary_date) + ON CONFLICT (test_id, prow_job_id, suite_id, release, %s) DO UPDATE SET %s WHERE (%s) IS DISTINCT FROM (%s)`, + dateColumn, strings.Join(setClauses, ", "), strings.Join(oldCols, ", "), strings.Join(newCols, ", ")) } type summaryStore interface { - MaxSummaryDate() (*time.Time, error) - Truncate() error + MaxSummaryDate() (*civil.Date, error) Releases() ([]string, error) - AggregateRangeForRelease(start, end time.Time, release string, skipConflictDetection bool) error -} - -// Options configures the daily summary refresh. -type Options struct { - Rebuild bool - StartOverride *time.Time - EndOverride *time.Time + AggregateRangeForRelease(start, end civil.Date, release string, skipConflictDetection bool) error } -// Refresh aggregates prow_job_run_tests into the test_daily_summaries -// table. It runs before matview refreshes so the matviews read from -// pre-aggregated data instead of scanning raw rows. -func Refresh(dbc *db.DB, opts Options) error { - return refreshSummaries(&pgStore{dbc: dbc}, opts) +// Refresh aggregates prow_job_run_tests into test_daily_summaries. +// Returns the earliest date that was refreshed so downstream consumers +// (cumulative summaries) know which dates may have changed. +func Refresh(dbc *db.DB) (civil.Date, error) { + return refreshSummaries(&pgStore{dbc: dbc, tableName: "test_daily_summaries", dateColumn: "summary_date"}) } -func refreshSummaries(store summaryStore, opts Options) error { +func refreshSummaries(store summaryStore) (civil.Date, error) { loadStart := time.Now() log.Info("refreshing daily summaries") - now := time.Now() - - startDate, endDate, err := dateRange(store, opts, now) + today := civil.DateOf(time.Now().UTC()) + maxDate, err := store.MaxSummaryDate() if err != nil { - return err + return civil.Date{}, fmt.Errorf("querying max summary date: %w", err) + } + + startDate := startDateFromMax(maxDate, today) + skipConflictDetection := maxDate == nil + + if err := doAggregate(store, startDate, today, skipConflictDetection, loadStart); err != nil { + return civil.Date{}, err } + return startDate, nil +} + +// Backfill processes an explicit date range without automatic date detection. +func Backfill(dbc *db.DB, startDate, endDate civil.Date) error { + return backfillSummaries(&pgStore{dbc: dbc, tableName: "test_daily_summaries", dateColumn: "summary_date"}, startDate, endDate) +} + +// RefreshTotals aggregates prow_job_run_tests into the partitioned +// test_daily_totals table. Same logic as Refresh but targets the new table. +func RefreshTotals(dbc *db.DB) (civil.Date, error) { + return refreshSummaries(&pgStore{dbc: dbc, tableName: "test_daily_totals", dateColumn: "date"}) +} + +// BackfillTotals backfills the partitioned test_daily_totals table. +func BackfillTotals(dbc *db.DB, startDate, endDate civil.Date) error { + return backfillSummaries(&pgStore{dbc: dbc, tableName: "test_daily_totals", dateColumn: "date"}, startDate, endDate) +} +func backfillSummaries(store summaryStore, startDate, endDate civil.Date) error { + loadStart := time.Now() releases, err := store.Releases() if err != nil { return fmt.Errorf("querying releases: %w", err) } - skipConflictDetection := opts.Rebuild - if skipConflictDetection { - log.Info("rebuild requested, truncating test_daily_summaries") - if err := store.Truncate(); err != nil { - return fmt.Errorf("truncating table: %w", err) - } - } else { - maxDate, err := store.MaxSummaryDate() - if err != nil { - return fmt.Errorf("checking if table is empty: %w", err) + days := endDate.DaysSince(startDate) + 1 + log.WithFields(log.Fields{ + "start": startDate, + "end": endDate, + "days": days, + "releases": len(releases), + }).Info("backfilling daily summaries") + + for date := startDate; !date.After(endDate); date = date.AddDays(1) { + dayStart := time.Now() + if err := aggregateReleases(store, releases, date, date, false); err != nil { + return fmt.Errorf("backfilling %s: %w", date, err) } - skipConflictDetection = maxDate == nil + log.WithFields(log.Fields{ + "date": date, + "elapsed": time.Since(dayStart), + }).Info("backfilled daily summaries for date") } - log.Infof("aggregating daily summaries from %s to %s", - startDate.Format("2006-01-02"), endDate.Format("2006-01-02")) + log.WithField("elapsed", time.Since(loadStart)).Info("daily summary backfill complete") + return nil +} - if err := aggregateReleases(store, releases, startDate, endDate, skipConflictDetection); err != nil { - return err +func doAggregate(store summaryStore, startDate, endDate civil.Date, skipConflictDetection bool, loadStart time.Time) error { + releases, err := store.Releases() + if err != nil { + return fmt.Errorf("querying releases: %w", err) + } + + days := endDate.DaysSince(startDate) + 1 + log.WithFields(log.Fields{ + "start": startDate, + "end": endDate, + "days": days, + "releases": len(releases), + }).Info("aggregating daily summaries") + + for date := startDate; !date.After(endDate); date = date.AddDays(1) { + dayStart := time.Now() + if err := aggregateReleases(store, releases, date, date, skipConflictDetection); err != nil { + return fmt.Errorf("aggregating %s: %w", date, err) + } + log.WithFields(log.Fields{ + "date": date, + "elapsed": time.Since(dayStart), + }).Debug("aggregated daily summaries for date") } log.WithField("elapsed", time.Since(loadStart)).Info("daily summary refresh complete") return nil } -func aggregateReleases(store summaryStore, releases []string, startDate, endDate time.Time, skipConflictDetection bool) error { +func aggregateReleases(store summaryStore, releases []string, startDate, endDate civil.Date, skipConflictDetection bool) error { errs := make(chan error, len(releases)) work := make(chan string, len(releases)) @@ -155,81 +197,46 @@ func aggregateReleases(store summaryStore, releases []string, startDate, endDate return errors.Join(combined...) } -// dateRange computes the aggregation window. If explicit overrides were -// provided, those are used directly. Otherwise the start is the last -// summarized date (capped at yesterday) or the default lookback, -// and the end is now. -func dateRange(store summaryStore, opts Options, now time.Time) (time.Time, time.Time, error) { - if opts.StartOverride != nil && opts.EndOverride != nil { - return *opts.StartOverride, *opts.EndOverride, nil - } - - endDate := now - if opts.EndOverride != nil { - endDate = *opts.EndOverride - } - - if opts.StartOverride != nil { - return *opts.StartOverride, endDate, nil - } - - if opts.Rebuild { - return now.AddDate(0, 0, -defaultLookbackDays), endDate, nil - } - - startDate, err := resolveStartDate(store, now) - if err != nil { - return time.Time{}, time.Time{}, fmt.Errorf("querying max summary date: %w", err) - } - - return startDate, endDate, nil -} - -// resolveStartDate returns the last summarized date capped at yesterday, -// or the default lookback if no summaries exist. -func resolveStartDate(store summaryStore, now time.Time) (time.Time, error) { - yesterday := now.AddDate(0, 0, -1) - - maxSummary, err := store.MaxSummaryDate() - if err != nil { - return time.Time{}, err - } +func startDateFromMax(maxSummary *civil.Date, today civil.Date) civil.Date { + yesterday := today.AddDays(-1) if maxSummary != nil { if maxSummary.Before(yesterday) { - return *maxSummary, nil + return *maxSummary } - return yesterday, nil + return yesterday } - - return now.AddDate(0, 0, -defaultLookbackDays), nil + return today.AddDays(-defaultLookbackDays) } -// pgStore implements summaryStore against PostgreSQL. type pgStore struct { - dbc *db.DB + dbc *db.DB + tableName string + dateColumn string } -func (s *pgStore) MaxSummaryDate() (*time.Time, error) { - var maxDate *time.Time - err := s.dbc.DB.Table("test_daily_summaries"). - Select("MAX(summary_date)").Row().Scan(&maxDate) - return maxDate, err -} +func (s *pgStore) MaxSummaryDate() (*civil.Date, error) { + var d *civil.Date + err := s.dbc.DB.Table(s.tableName). + Select(fmt.Sprintf("MAX(%s)", s.dateColumn)).Row().Scan(&d) + if err != nil { + return nil, err + } -func (s *pgStore) Truncate() error { - return s.dbc.DB.Exec("TRUNCATE test_daily_summaries").Error + return d, nil } func (s *pgStore) Releases() ([]string, error) { var releases []string - err := s.dbc.DB.Table("prow_jobs").Distinct("release").Pluck("release", &releases).Error + err := s.dbc.DB.Table("release_definitions"). + Order("major DESC, minor DESC"). + Pluck("release", &releases).Error return releases, err } -func (s *pgStore) AggregateRangeForRelease(startDate, endDate time.Time, release string, skipConflictDetection bool) error { - sql := insertSQL +func (s *pgStore) AggregateRangeForRelease(startDate, endDate civil.Date, release string, skipConflictDetection bool) error { + sql := buildInsertSQL(s.tableName, s.dateColumn) if !skipConflictDetection { - sql += onConflictClause + sql += buildOnConflictClause(s.tableName, s.dateColumn) } return s.dbc.DB.Exec(sql, startDate, endDate, release).Error } diff --git a/pkg/db/dailysummary/dailysummary_test.go b/pkg/db/dailysummary/dailysummary_test.go index 2f0adf6a9f..68958ae512 100644 --- a/pkg/db/dailysummary/dailysummary_test.go +++ b/pkg/db/dailysummary/dailysummary_test.go @@ -6,15 +6,15 @@ import ( "testing" "time" + "cloud.google.com/go/civil" "github.com/stretchr/testify/assert" "github.com/stretchr/testify/require" + "k8s.io/apimachinery/pkg/util/sets" ) type fakeStore struct { - maxSummary *time.Time + maxSummary *civil.Date maxSummaryErr error - truncated bool - truncateErr error releases []string releasesErr error aggregateErr error @@ -24,20 +24,15 @@ type fakeStore struct { } type aggregateCall struct { - start, end time.Time + start, end civil.Date release string skipConflictDetection bool } -func (f *fakeStore) MaxSummaryDate() (*time.Time, error) { +func (f *fakeStore) MaxSummaryDate() (*civil.Date, error) { return f.maxSummary, f.maxSummaryErr } -func (f *fakeStore) Truncate() error { - f.truncated = true - return f.truncateErr -} - func (f *fakeStore) Releases() ([]string, error) { if f.releases != nil { return f.releases, f.releasesErr @@ -45,7 +40,7 @@ func (f *fakeStore) Releases() ([]string, error) { return []string{"4.22", "5.0"}, f.releasesErr } -func (f *fakeStore) AggregateRangeForRelease(start, end time.Time, release string, skipConflictDetection bool) error { +func (f *fakeStore) AggregateRangeForRelease(start, end civil.Date, release string, skipConflictDetection bool) error { f.mu.Lock() defer f.mu.Unlock() f.calls = append(f.calls, aggregateCall{start: start, end: end, release: release, skipConflictDetection: skipConflictDetection}) @@ -83,106 +78,81 @@ func (f *fakeStore) noneSkippedConflictDetection() bool { } func TestRefresh_Incremental(t *testing.T) { - maxDate := time.Date(2026, 6, 17, 0, 0, 0, 0, time.UTC) + today := civil.DateOf(time.Now().UTC()) + maxDate := today.AddDays(-3) store := &fakeStore{maxSummary: &maxDate} - err := refreshSummaries(store, Options{}) + _, err := refreshSummaries(store) require.NoError(t, err) - assert.Len(t, store.getCalls(), 2) + // 4 days (maxDate through today) × 2 releases + assert.Len(t, store.getCalls(), 4*2) assert.True(t, store.noneSkippedConflictDetection()) - assert.False(t, store.truncated) } -func TestRefresh_IncrementalCapsAtYesterday(t *testing.T) { - future := time.Now().AddDate(0, 0, 1) +func TestRefresh_FutureMaxDateStartsAtYesterday(t *testing.T) { + future := civil.Date{Year: 2099, Month: 1, Day: 1} store := &fakeStore{maxSummary: &future} - err := refreshSummaries(store, Options{}) + _, err := refreshSummaries(store) require.NoError(t, err) calls := store.getCalls() require.NotEmpty(t, calls) - assert.True(t, calls[0].start.Before(future)) + today := civil.DateOf(time.Now().UTC()) + yesterday := today.AddDays(-1) + assert.Equal(t, yesterday, calls[0].start) + assert.Equal(t, today, calls[len(calls)-1].end) } func TestRefresh_EmptyTableUsesDefaultLookbackAndSkipsConflictDetection(t *testing.T) { store := &fakeStore{} - err := refreshSummaries(store, Options{}) + _, err := refreshSummaries(store) require.NoError(t, err) calls := store.getCalls() require.NotEmpty(t, calls) - daysBefore := time.Since(calls[0].start).Hours() / 24 + today := civil.DateOf(time.Now().UTC()) + daysBefore := today.DaysSince(calls[0].start) assert.InDelta(t, defaultLookbackDays, daysBefore, 1) assert.True(t, store.allSkippedConflictDetection()) } -func TestRefresh_RebuildUsesInsert(t *testing.T) { - store := &fakeStore{} - - err := refreshSummaries(store, Options{Rebuild: true}) - - require.NoError(t, err) - assert.True(t, store.truncated) - assert.Len(t, store.getCalls(), 2) - assert.True(t, store.allSkippedConflictDetection()) -} - func TestRefresh_IncrementalUsesUpsert(t *testing.T) { - maxDate := time.Date(2026, 6, 17, 0, 0, 0, 0, time.UTC) + today := civil.DateOf(time.Now().UTC()) + maxDate := today.AddDays(-2) store := &fakeStore{maxSummary: &maxDate} - err := refreshSummaries(store, Options{}) + _, err := refreshSummaries(store) require.NoError(t, err) - assert.Len(t, store.getCalls(), 2) + // 3 days × 2 releases + assert.Len(t, store.getCalls(), 3*2) assert.True(t, store.noneSkippedConflictDetection()) } -func TestRefresh_DateOverrides(t *testing.T) { - start := time.Date(2026, 1, 1, 0, 0, 0, 0, time.UTC) - end := time.Date(2026, 3, 1, 0, 0, 0, 0, time.UTC) - maxDate := time.Date(2026, 6, 17, 0, 0, 0, 0, time.UTC) - store := &fakeStore{maxSummary: &maxDate} - - err := refreshSummaries(store, Options{StartOverride: &start, EndOverride: &end}) - - require.NoError(t, err) - calls := store.getCalls() - require.Len(t, calls, 2) - assert.Equal(t, start, calls[0].start) - assert.Equal(t, end, calls[0].end) -} - -func TestRefresh_StartOverrideOnly(t *testing.T) { - start := time.Date(2026, 1, 1, 0, 0, 0, 0, time.UTC) - maxDate := time.Date(2026, 6, 17, 0, 0, 0, 0, time.UTC) - store := &fakeStore{maxSummary: &maxDate} +func TestBackfill_UsesExplicitDateRange(t *testing.T) { + start := civil.Date{Year: 2026, Month: 7, Day: 1} + end := civil.Date{Year: 2026, Month: 7, Day: 3} + store := &fakeStore{} - err := refreshSummaries(store, Options{StartOverride: &start}) + err := backfillSummaries(store, start, end) require.NoError(t, err) calls := store.getCalls() - require.Len(t, calls, 2) + // 3 days × 2 releases = 6 calls, each single-day + require.Len(t, calls, 6) assert.Equal(t, start, calls[0].start) -} - -func TestRefresh_TruncateError(t *testing.T) { - store := &fakeStore{truncateErr: fmt.Errorf("permission denied")} - - err := refreshSummaries(store, Options{Rebuild: true}) - - require.Error(t, err) - assert.Contains(t, err.Error(), "truncating table") - assert.Empty(t, store.getCalls()) + assert.Equal(t, start, calls[0].end) + assert.Equal(t, end, calls[len(calls)-1].start) + assert.Equal(t, end, calls[len(calls)-1].end) } func TestRefresh_MaxSummaryDateError(t *testing.T) { store := &fakeStore{maxSummaryErr: fmt.Errorf("connection refused")} - err := refreshSummaries(store, Options{}) + _, err := refreshSummaries(store) require.Error(t, err) assert.Contains(t, err.Error(), "max summary date") @@ -192,7 +162,7 @@ func TestRefresh_MaxSummaryDateError(t *testing.T) { func TestRefresh_ReleasesError(t *testing.T) { store := &fakeStore{releasesErr: fmt.Errorf("connection refused")} - err := refreshSummaries(store, Options{}) + _, err := refreshSummaries(store) require.Error(t, err) assert.Contains(t, err.Error(), "querying releases") @@ -200,10 +170,10 @@ func TestRefresh_ReleasesError(t *testing.T) { } func TestRefresh_AggregateError(t *testing.T) { - maxDate := time.Date(2026, 6, 17, 0, 0, 0, 0, time.UTC) + maxDate := civil.Date{Year: 2026, Month: 6, Day: 17} store := &fakeStore{maxSummary: &maxDate, aggregateErr: fmt.Errorf("disk full")} - err := refreshSummaries(store, Options{}) + _, err := refreshSummaries(store) require.Error(t, err) assert.Contains(t, err.Error(), "aggregating release") @@ -212,27 +182,30 @@ func TestRefresh_AggregateError(t *testing.T) { func TestRefresh_NoReleases(t *testing.T) { store := &fakeStore{releases: []string{}} - err := refreshSummaries(store, Options{}) + _, err := refreshSummaries(store) require.NoError(t, err) assert.Empty(t, store.getCalls()) } -func TestRefresh_ParallelProcessesAllReleases(t *testing.T) { +func TestRefresh_ProcessesAllReleasesPerDay(t *testing.T) { releases := []string{"4.18", "4.19", "4.20", "4.21", "4.22", "5.0"} - maxDate := time.Date(2026, 6, 17, 0, 0, 0, 0, time.UTC) + today := civil.DateOf(time.Now().UTC()) + maxDate := today.AddDays(-3) store := &fakeStore{maxSummary: &maxDate, releases: releases} - err := refreshSummaries(store, Options{}) + _, err := refreshSummaries(store) require.NoError(t, err) calls := store.getCalls() - assert.Len(t, calls, len(releases)) - seen := make(map[string]bool) + // 4 days (maxDate through today) × 6 releases = 24 calls + assert.Len(t, calls, 4*len(releases)) + seen := sets.New[string]() for _, call := range calls { - seen[call.release] = true + seen.Insert(call.release) + assert.Equal(t, call.start, call.end, "each call should be a single day") } for _, rel := range releases { - assert.True(t, seen[rel], "release %s not processed", rel) + assert.True(t, seen.Has(rel), "release %s not processed", rel) } } diff --git a/pkg/db/db.go b/pkg/db/db.go index d9820d0819..d80ebec8b9 100644 --- a/pkg/db/db.go +++ b/pkg/db/db.go @@ -29,6 +29,8 @@ const ( partitionedTableProwJobRunTests = "prow_job_run_tests" partitionedTableProwJobRunTestsOutputs = "prow_job_run_test_outputs" partitionedTableTestAnalysisByJobByDates = "test_analysis_by_job_by_dates" + partitionedTableTestDailyTotals = "test_daily_totals" + partitionedTableTestCumulativeSummaries = "test_cumulative_summaries" ) type DB struct { @@ -89,6 +91,8 @@ func New(dsn string, logLevel gormlogger.LogLevel, opts ...Option) (*DB, error) // partitions. Custom plans use actual parameter values for partition pruning. pgxConfig.RuntimeParams["plan_cache_mode"] = "force_custom_plan" pgxConfig.RuntimeParams["work_mem"] = "128MB" + pgxConfig.RuntimeParams["idle_in_transaction_session_timeout"] = "60s" + pgxConfig.RuntimeParams["random_page_cost"] = "1.1" if cfg.enablePartitionwise { pgxConfig.RuntimeParams["enable_partitionwise_aggregate"] = "on" pgxConfig.RuntimeParams["enable_partitionwise_join"] = "on" @@ -216,6 +220,8 @@ func (d *DB) PartitionedTables() []string { partitionedTableProwJobRunTests, partitionedTableProwJobRunTestsOutputs, partitionedTableTestAnalysisByJobByDates, + partitionedTableTestDailyTotals, + partitionedTableTestCumulativeSummaries, } } @@ -241,7 +247,7 @@ func (d *DB) EnsurePartitions(releases []string, startDate, endDate time.Time, d dateColumn = "prow_job_run_timestamp" case partitionedTableProwJobRunTestsOutputs: dateColumn = "prow_job_run_test_timestamp" - case partitionedTableTestAnalysisByJobByDates: + case partitionedTableTestAnalysisByJobByDates, partitionedTableTestDailyTotals, partitionedTableTestCumulativeSummaries: dateColumn = "date" default: log.Warnf("unknown partitioned table: %s", tableName) diff --git a/pkg/db/migrations/000006_partition_cumulative_tables.down.sql b/pkg/db/migrations/000006_partition_cumulative_tables.down.sql new file mode 100644 index 0000000000..6b2438a247 --- /dev/null +++ b/pkg/db/migrations/000006_partition_cumulative_tables.down.sql @@ -0,0 +1,2 @@ +DROP TABLE IF EXISTS test_cumulative_summaries CASCADE; +DROP TABLE IF EXISTS test_daily_totals CASCADE; diff --git a/pkg/db/migrations/000006_partition_cumulative_tables.up.sql b/pkg/db/migrations/000006_partition_cumulative_tables.up.sql new file mode 100644 index 0000000000..cee4a021e3 --- /dev/null +++ b/pkg/db/migrations/000006_partition_cumulative_tables.up.sql @@ -0,0 +1,55 @@ +-- TRT-2741: Create partitioned summary tables +-- +-- Creates 2 partitioned tables using nested LIST->RANGE partitioning: +-- - Level 1: LIST partition by release +-- - Level 2: RANGE sub-partition by date (daily granularity) +-- +-- Tables: +-- - test_daily_totals (partitioned daily aggregates, replaces test_daily_summaries) +-- - test_cumulative_summaries (prefix sums from test_daily_totals) +-- +-- Partition creation (release partitions + daily sub-partitions) is handled +-- by the partition management system (gopar). + +-- ============================================================================ +-- 1. test_daily_totals +-- ============================================================================ + +CREATE TABLE IF NOT EXISTS test_daily_totals ( + test_id BIGINT NOT NULL, + prow_job_id BIGINT NOT NULL, + suite_id BIGINT NOT NULL DEFAULT 0, + release TEXT NOT NULL, + date DATE NOT NULL, + successes INT NOT NULL DEFAULT 0, + failures INT NOT NULL DEFAULT 0, + flakes INT NOT NULL DEFAULT 0, + runs INT NOT NULL DEFAULT 0 +) PARTITION BY LIST (release); + +CREATE UNIQUE INDEX IF NOT EXISTS idx_test_daily_totals_unique + ON test_daily_totals (test_id, prow_job_id, suite_id, release, date); + +CREATE INDEX IF NOT EXISTS idx_test_daily_totals_date + ON test_daily_totals (date); + +-- ============================================================================ +-- 2. test_cumulative_summaries +-- ============================================================================ + +CREATE TABLE IF NOT EXISTS test_cumulative_summaries ( + date DATE NOT NULL, + release TEXT NOT NULL, + test_id BIGINT NOT NULL, + prow_job_id BIGINT NOT NULL, + suite_id BIGINT NOT NULL DEFAULT 0, + prefix_sum_successes BIGINT NOT NULL DEFAULT 0, + prefix_sum_failures BIGINT NOT NULL DEFAULT 0, + prefix_sum_flakes BIGINT NOT NULL DEFAULT 0, + prefix_sum_runs BIGINT NOT NULL DEFAULT 0, + + PRIMARY KEY (date, release, test_id, prow_job_id, suite_id) +) PARTITION BY LIST (release); + +CREATE INDEX IF NOT EXISTS idx_test_cumulative_summaries_prow_job_id + ON test_cumulative_summaries (prow_job_id); diff --git a/pkg/db/migrations/MANIFEST b/pkg/db/migrations/MANIFEST index 3b73f52270..5e259ccf66 100644 --- a/pkg/db/migrations/MANIFEST +++ b/pkg/db/migrations/MANIFEST @@ -12,3 +12,4 @@ 000003_create_variant_combinations 000004_drop_variants_gin_index 000005_drop_daily_summary_variant_combination +000006_partition_cumulative_tables diff --git a/pkg/db/models/prow.go b/pkg/db/models/prow.go index f5c520dd6a..a8de3da835 100644 --- a/pkg/db/models/prow.go +++ b/pkg/db/models/prow.go @@ -3,6 +3,7 @@ package models import ( "time" + "cloud.google.com/go/civil" "github.com/lib/pq" "gorm.io/gorm" @@ -183,6 +184,41 @@ type TestDailySummary struct { Runs int32 `gorm:"column:runs;not null;default:0"` } +// TestDailyTotal is the partitioned replacement for TestDailySummary. +// Same schema, but partitioned by LIST(release) then RANGE(date). +// Table is partitioned (LIST by release, RANGE by date) - +// schema managed by migration 000006, not AutoMigrate. +type TestDailyTotal struct { + TestID uint `gorm:"column:test_id;not null"` + ProwJobID uint `gorm:"column:prow_job_id;not null"` + SuiteID uint `gorm:"column:suite_id;not null;default:0"` + Release string `gorm:"column:release;not null"` + Date civil.Date `gorm:"column:date;type:date;not null"` + Successes int32 `gorm:"column:successes;not null;default:0"` + Failures int32 `gorm:"column:failures;not null;default:0"` + Flakes int32 `gorm:"column:flakes;not null;default:0"` + Runs int32 `gorm:"column:runs;not null;default:0"` +} + +// TestCumulativeSummary stores running totals of test_daily_summaries values, +// ordered by date. Any date range [start, end] can be computed as +// cumulative(end) - cumulative(start-1). Keyed by immutable fields only +// (no variant_combination_id) so variant changes do not invalidate the data. +// Entities are carried forward on days with no data so the chain is unbroken. +// Table is partitioned (LIST by release, RANGE by date) - +// schema managed by migration 000006, not AutoMigrate. +type TestCumulativeSummary struct { + Date civil.Date `gorm:"column:date;type:date;not null;primaryKey;priority:1"` + Release string `gorm:"column:release;not null;primaryKey;priority:2"` + TestID uint `gorm:"column:test_id;not null;primaryKey;priority:3"` + ProwJobID uint `gorm:"column:prow_job_id;not null;primaryKey;priority:4;index:idx_test_cumulative_summaries_prow_job_id"` + SuiteID uint `gorm:"column:suite_id;not null;default:0;primaryKey;priority:5"` + PrefixSumSuccesses int64 `gorm:"column:prefix_sum_successes;not null;default:0"` + PrefixSumFailures int64 `gorm:"column:prefix_sum_failures;not null;default:0"` + PrefixSumFlakes int64 `gorm:"column:prefix_sum_flakes;not null;default:0"` + PrefixSumRuns int64 `gorm:"column:prefix_sum_runs;not null;default:0"` +} + // Bug represents a Jira bug. type Bug struct { ID uint `json:"id" gorm:"primaryKey"` diff --git a/pkg/flags/postgres_benchmarking_test.go b/pkg/flags/postgres_benchmarking_test.go index 5767188ee4..dee1ed5662 100644 --- a/pkg/flags/postgres_benchmarking_test.go +++ b/pkg/flags/postgres_benchmarking_test.go @@ -13,7 +13,6 @@ import ( apitype "github.com/openshift/sippy/pkg/apis/api" v1 "github.com/openshift/sippy/pkg/apis/sippyprocessing/v1" "github.com/openshift/sippy/pkg/db" - "github.com/openshift/sippy/pkg/db/dailysummary" "github.com/openshift/sippy/pkg/db/models" "github.com/openshift/sippy/pkg/db/query" "github.com/openshift/sippy/pkg/filter" @@ -1231,8 +1230,7 @@ func Test_BenchmarkRefreshData(t *testing.T) { r := runBenchmarkCase(t, dbc, benchmarkCase{ name: "RefreshData", fn: func(dbc *db.DB) error { - sippyserver.RefreshData(dbc, nil, false, dailysummary.Options{}) - return nil + return sippyserver.RefreshData(dbc, nil, sippyserver.RefreshOptions{}) }, }, 1) printSummaryTable(t, []benchmarkResult{r}, connName) diff --git a/pkg/sippyserver/server.go b/pkg/sippyserver/server.go index cc98a289d8..03d74818c0 100644 --- a/pkg/sippyserver/server.go +++ b/pkg/sippyserver/server.go @@ -45,6 +45,7 @@ import ( "github.com/andygrunwald/go-jira" + "cloud.google.com/go/civil" "github.com/openshift/sippy/pkg/api" "github.com/openshift/sippy/pkg/api/componentreadiness" "github.com/openshift/sippy/pkg/api/jobrunevents" @@ -54,6 +55,7 @@ import ( sippyv1 "github.com/openshift/sippy/pkg/apis/sippy/v1" sippybq "github.com/openshift/sippy/pkg/bigquery" "github.com/openshift/sippy/pkg/db" + "github.com/openshift/sippy/pkg/db/cumulativesummary" "github.com/openshift/sippy/pkg/db/dailysummary" "github.com/openshift/sippy/pkg/db/models" "github.com/openshift/sippy/pkg/db/query" @@ -151,6 +153,12 @@ var dailySummaryRefreshMetric = promauto.NewHistogram(prometheus.HistogramOpts{ Buckets: []float64{1000, 5000, 10000, 30000, 60000, 300000, 600000, 1200000}, }) +var cumulativeSummaryRefreshMetric = promauto.NewHistogram(prometheus.HistogramOpts{ + Name: "sippy_cumulative_summary_refresh_millis", + Help: "Milliseconds to refresh the cumulative summary tables", + Buckets: []float64{100, 500, 1000, 5000, 10000, 30000, 60000, 300000, 600000}, +}) + var matViewUniqueNumberOfTests = promauto.NewGaugeVec(prometheus.GaugeOpts{ Name: "sippy_matviews_unique_number_of_tests", Help: "Total number of tests based on lookback days", @@ -387,16 +395,59 @@ func recordMatviewRefreshTime(cacheClient cache.Cache, matView string, tmpLog *l } } -func RefreshData(dbc *db.DB, cacheClient cache.Cache, refreshMatviewsOnlyIfEmpty bool, dailySummaryOpts dailysummary.Options) { +// RefreshOptions controls the incremental refresh behavior. +type RefreshOptions struct { + RefreshOnlyIfEmpty bool +} + +// RefreshData runs the normal incremental refresh of all summary tables +// and materialized views. Used by the load command and the serve loop. +func RefreshData(dbc *db.DB, cacheClient cache.Cache, opts RefreshOptions) error { log.Infof("Refreshing data") + summaryStart := time.Now() - if err := dailysummary.Refresh(dbc, dailySummaryOpts); err != nil { - log.WithError(err).Error("failed to refresh daily summaries") - } else { - dailySummaryRefreshMetric.Observe(float64(time.Since(summaryStart).Milliseconds())) + if _, err := dailysummary.Refresh(dbc); err != nil { + return fmt.Errorf("failed to refresh daily summaries: %w", err) + } + dailySummaryRefreshMetric.Observe(float64(time.Since(summaryStart).Milliseconds())) + + totalsStart := time.Now() + earliestTotalsChanged, err := dailysummary.RefreshTotals(dbc) + if err != nil { + return fmt.Errorf("failed to refresh daily totals: %w", err) } - refreshMaterializedViews(dbc, cacheClient, refreshMatviewsOnlyIfEmpty) + log.WithField("elapsed", time.Since(totalsStart)).Info("daily totals refresh complete") + + cumulativeStart := time.Now() + if _, err := cumulativesummary.Refresh(dbc, earliestTotalsChanged); err != nil { + return fmt.Errorf("failed to refresh cumulative summaries: %w", err) + } + cumulativeSummaryRefreshMetric.Observe(float64(time.Since(cumulativeStart).Milliseconds())) + + refreshMaterializedViews(dbc, cacheClient, opts.RefreshOnlyIfEmpty) log.Info("Refresh complete") + return nil +} + +// BackfillData refreshes a specific table for a given date range, +// bypassing the normal incremental logic. Used for backfilling. +func BackfillData(dbc *db.DB, table string, startDate, endDate civil.Date) error { + log.WithFields(log.Fields{ + "table": table, + "start": startDate, + "end": endDate, + }).Info("Backfilling table") + + switch table { + case "daily-summaries": + return dailysummary.Backfill(dbc, startDate, endDate) + case "daily-totals": + return dailysummary.BackfillTotals(dbc, startDate, endDate) + case "cumulative-summaries": + return cumulativesummary.Backfill(dbc, startDate, endDate) + default: + return fmt.Errorf("unknown table: %s", table) + } } func (s *Server) hasCapabilities(capabilities []string) bool { diff --git a/scripts/backfill-summaries.sh b/scripts/backfill-summaries.sh new file mode 100755 index 0000000000..049577eea6 --- /dev/null +++ b/scripts/backfill-summaries.sh @@ -0,0 +1,141 @@ +#!/bin/bash +# Gradually backfill a summary table one day at a time. +# Each day runs as a separate pod to avoid long-running transactions. +# +# Usage: +# ./scripts/backfill-summaries.sh --table TABLE [OPTIONS] +# +# Options: +# --table TABLE Table to backfill (daily-summaries, daily-totals, +# cumulative-summaries) [required] +# --days N Number of days to backfill (default: 91) +# --namespace NS Kubernetes namespace (default: sippy) +# --image IMAGE Container image (default: auto-detect from sippy DC) +# --db-secret NAME Database secret name [required] +# --pause SECONDS Seconds to wait between days (default: 5) +# --batch N Days per pod (default: 1) +# --dry-run Print commands without executing + +set -euo pipefail + +DAYS=91 +TABLE="" +NAMESPACE=sippy +IMAGE="" +DB_SECRET="" +PAUSE=5 +BATCH=1 +DRY_RUN=false + +while [[ $# -gt 0 ]]; do + case "$1" in + --table) TABLE="$2"; shift 2 ;; + --days) DAYS="$2"; shift 2 ;; + --namespace) NAMESPACE="$2"; shift 2 ;; + --image) IMAGE="$2"; shift 2 ;; + --db-secret) DB_SECRET="$2"; shift 2 ;; + --pause) PAUSE="$2"; shift 2 ;; + --batch) BATCH="$2"; shift 2 ;; + --dry-run) DRY_RUN=true; shift ;; + *) echo "Unknown option: $1" >&2; exit 1 ;; + esac +done + +for arg_name in DAYS PAUSE BATCH; do + arg_val="${!arg_name}" + if ! [[ "$arg_val" =~ ^[1-9][0-9]*$ ]]; then + echo "Error: --$(echo "$arg_name" | tr '[:upper:]' '[:lower:]') must be a positive integer, got '$arg_val'" >&2 + exit 1 + fi +done + +if [[ -z "$DB_SECRET" ]]; then + echo "Error: --db-secret is required" >&2 + exit 1 +fi + +if [[ -z "$TABLE" ]]; then + echo "Error: --table is required" >&2 + echo "Valid tables: daily-summaries, daily-totals, cumulative-summaries" >&2 + exit 1 +fi + +case "$TABLE" in + daily-summaries|daily-totals|cumulative-summaries) ;; + *) + echo "Error: invalid table '$TABLE'" >&2 + echo "Valid tables: daily-summaries, daily-totals, cumulative-summaries" >&2 + exit 1 + ;; +esac + +if [[ -z "$IMAGE" ]]; then + IMAGE=$(oc -n "$NAMESPACE" get dc/sippy -o jsonpath='{.spec.template.spec.containers[0].image}' 2>/dev/null) || true + if [[ -z "$IMAGE" ]]; then + echo "Could not auto-detect image. Use --image to specify." >&2 + exit 1 + fi + echo "Using image: $IMAGE" +fi + +today=$(date -u +%Y-%m-%d) +completed=0 +failed=0 + +for ((offset=DAYS; offset>0; offset-=BATCH)); do + batch_end_offset=$((offset - BATCH + 1)) + if (( batch_end_offset < 0 )); then + batch_end_offset=0 + fi + + start_date=$(date -u -v-${offset}d +%Y-%m-%d 2>/dev/null || date -u -d "$today - ${offset} days" +%Y-%m-%d) + end_date=$(date -u -v-${batch_end_offset}d +%Y-%m-%d 2>/dev/null || date -u -d "$today - ${batch_end_offset} days" +%Y-%m-%d) + + pod_name="sippy-backfill-$(echo "$start_date" | tr -d '-')" + echo "[$(date +%H:%M:%S)] Processing $start_date to $end_date ($((DAYS - offset + BATCH))/$DAYS days)..." + + if [[ "$DRY_RUN" == "true" ]]; then + echo " Would create pod $pod_name: sippy backfill --table=$TABLE --start-date=$start_date --end-date=$end_date" + completed=$((completed + 1)) + continue + fi + + # Clean up any leftover pod from a previous run + oc -n "$NAMESPACE" delete pod "$pod_name" --ignore-not-found --wait=false >/dev/null 2>&1 || true + sleep 2 + + oc -n "$NAMESPACE" run "$pod_name" --rm -i --restart=Never \ + --image="$IMAGE" \ + --overrides="{ + \"spec\": { + \"containers\": [{ + \"name\": \"$pod_name\", + \"image\": \"$IMAGE\", + \"command\": [\"sippy\"], + \"args\": [\"backfill\", + \"--table=$TABLE\", + \"--start-date=$start_date\", + \"--end-date=$end_date\"], + \"env\": [{ + \"name\": \"SIPPY_DATABASE_DSN\", + \"valueFrom\": {\"secretKeyRef\": {\"name\": \"$DB_SECRET\", \"key\": \"dsn\"}} + }] + }] + } + }" < /dev/null 2>&1 | { grep -iE 'summar|error|elapsed|complete|panic|fatal|fail|timeout|warn' || true; } || true + exit_code=${PIPESTATUS[0]} + if [[ $exit_code -eq 0 ]]; then + completed=$((completed + 1)) + echo " Done." + else + failed=$((failed + 1)) + echo " FAILED (exit $exit_code)" >&2 + fi + + if (( offset - BATCH > 0 )); then + sleep "$PAUSE" + fi +done + +echo "" +echo "Backfill complete: $completed succeeded, $failed failed out of $(( (DAYS + BATCH - 1) / BATCH )) batches." diff --git a/scripts/updatesuites.go b/scripts/updatesuites.go index b5f8012778..3b8ffb3931 100644 --- a/scripts/updatesuites.go +++ b/scripts/updatesuites.go @@ -12,7 +12,6 @@ import ( gormlogger "gorm.io/gorm/logger" "github.com/openshift/sippy/pkg/db" - "github.com/openshift/sippy/pkg/db/dailysummary" "github.com/openshift/sippy/pkg/db/models" "github.com/openshift/sippy/pkg/sippyserver" ) @@ -152,9 +151,11 @@ func testNameWithoutSuite(dbc *gorm.DB) error { // Refresh materialized views // NOTE: does not update timestamps to invalidate cached matview data; not clear if the use case for this script requires that. - sippyserver.RefreshData(&db.DB{ + if err := sippyserver.RefreshData(&db.DB{ DB: dbc, - }, nil, false, dailysummary.Options{Rebuild: true}) + }, nil, sippyserver.RefreshOptions{}); err != nil { + return fmt.Errorf("refreshing data: %w", err) + } return nil }