TRT-2741: Add GA release test data loader sourced from BigQuery

cmd/sippy/load.go

@@ -32,6 +32,7 @@ import (
 	v1 "github.com/openshift/sippy/pkg/apis/config/v1"
 	"github.com/openshift/sippy/pkg/dataloader"
 	"github.com/openshift/sippy/pkg/dataloader/bugloader"
+	"github.com/openshift/sippy/pkg/dataloader/gateststatus"
 	"github.com/openshift/sippy/pkg/dataloader/jiraloader"
 	"github.com/openshift/sippy/pkg/dataloader/loaderwithmetrics"
 	"github.com/openshift/sippy/pkg/dataloader/prowloader"
@@ -66,6 +67,7 @@ type LoadFlags struct {
 	LogLevel                string
 	ProwLoadSince           string
 	SkipMatviewRefresh      bool
+	ForceGARefresh          bool
 }
 
 // want a single total load and refresh time
@@ -107,6 +109,7 @@ func (f *LoadFlags) BindFlags(fs *pflag.FlagSet) {
 	fs.StringVar(&f.LogLevel, "log-level", "info", "Log level")
 	fs.StringVar(&f.ProwLoadSince, "prow-load-since", "", "Override how far back to load prow jobs (e.g. 2024-01-15T00:00:00Z or 72h for 72 hours ago)")
 	fs.BoolVar(&f.SkipMatviewRefresh, "skip-matview-refresh", false, "Skip refreshing materialized views after loading")
+	fs.BoolVar(&f.ForceGARefresh, "force-ga-refresh", false, "Force re-population of GA test status data from BigQuery")
 }
 
 // nolint:gocyclo
@@ -357,6 +360,17 @@ func NewLoadCommand() *cobra.Command {
 					loaders = append(loaders, fgLoader)
 				}
 
+				if l == "ga-test-status" {
+					refreshMatviews = true
+					if bigqueryErr != nil {
+						return errors.Wrap(bigqueryErr, "CRITICAL error getting BigQuery client which prevents ga-test-status loading")
+					}
+					if dbErr != nil {
+						return errors.Wrap(dbErr, "CRITICAL error getting postgres client which prevents ga-test-status loading")
+					}
+					loaders = append(loaders, gateststatus.New(ctx, dbc, bqc, f.ForceGARefresh, f.Releases))
+				}
+
 			}
 
 			// Run loaders with the metrics wrapper

cmd/sippy/seed_data.go

@@ -3,6 +3,7 @@ package main
 import (
 	"context"
 	"database/sql"
+	stderrors "errors"
 	"fmt"
 	"os"
 	"sort"
@@ -16,6 +17,8 @@ import (
 	"github.com/spf13/cobra"
 	"github.com/spf13/pflag"
 	"gopkg.in/yaml.v3"
+	"gorm.io/gorm"
+	"k8s.io/apimachinery/pkg/util/sets"
 
 	componentreadiness "github.com/openshift/sippy/pkg/api/componentreadiness"
 	pgprovider "github.com/openshift/sippy/pkg/api/componentreadiness/dataprovider/postgres"
@@ -504,6 +507,12 @@ func seedSyntheticData(dbc *db.DB) error {
 			return fmt.Errorf("failed to backfill %s: %w", table, err)
 		}
 	}
+
+	if err := seedGARawTestData(dbc); err != nil {
+		return errors.WithMessage(err, "failed to seed GA raw test data")
+	}
+	log.Info("Seeded GA raw test data")
+
 	if err := sippyserver.RefreshData(dbc, nil, sippyserver.RefreshOptions{}); err != nil {
 		return fmt.Errorf("failed to refresh data: %w", err)
 	}
@@ -1045,3 +1054,114 @@ func seedFeatureGates(dbc *db.DB) error {
 	log.Infof("Created %d feature gate records", len(featureGates))
 	return nil
 }
+
+// seedGARawTestData populates prow_ga_raw_test_data for GA releases using
+// the same synthetic test/job definitions. This gives the
+// prow_ga_test_statuses_matview data to aggregate when refreshed.
+func seedGARawTestData(dbc *db.DB) error {
+	var gaReleases []models.ReleaseDefinition
+	if err := dbc.DB.Where("ga_date IS NOT NULL AND ga_date < CURRENT_DATE").Find(&gaReleases).Error; err != nil {
+		return fmt.Errorf("querying GA releases: %w", err)
+	}
+
+	if len(gaReleases) == 0 {
+		log.Info("No GA releases found, skipping GA raw test data seeding")
+		return nil
+	}
+
+	testIDCache := make(map[string]uint)
+	jobIDCache := make(map[string]uint)
+	var suiteID uint
+
+	var suite models.Suite
+	if err := dbc.DB.Where("name = ?", "synthetic").First(&suite).Error; err != nil {
+		if !stderrors.Is(err, gorm.ErrRecordNotFound) {
+			return fmt.Errorf("looking up suite 'synthetic': %w", err)
+		}
+		log.Warn("Suite 'synthetic' not found, GA raw test data will use suite_id=0")
+	} else {
+		suiteID = suite.ID
+	}
+
+	var rows []models.ProwGARawTestDatum
+	for _, rel := range gaReleases {
+		for _, windowDays := range utils.GAWindows {
+			for _, spec := range syntheticTests {
+				testID, ok := testIDCache[spec.testName]
+				if !ok {
+					var test models.Test
+					if err := dbc.DB.Where("name = ?", spec.testName).First(&test).Error; err != nil {
+						if !stderrors.Is(err, gorm.ErrRecordNotFound) {
+							return fmt.Errorf("looking up test %q: %w", spec.testName, err)
+						}
+						continue
+					}
+					testID = test.ID
+					testIDCache[spec.testName] = testID
+				}
+
+				for jobTemplate, releaseCounts := range spec.jobCounts {
+					counts, ok := releaseCounts[rel.Release]
+					if !ok {
+						continue
+					}
+					jobName := fmt.Sprintf(jobTemplate, rel.Release)
+					prowJobID, ok := jobIDCache[jobName]
+					if !ok {
+						var job models.ProwJob
+						if err := dbc.DB.Where("name = ?", jobName).First(&job).Error; err != nil {
+							if !stderrors.Is(err, gorm.ErrRecordNotFound) {
+								return fmt.Errorf("looking up prow job %q: %w", jobName, err)
+							}
+							continue
+						}
+						prowJobID = job.ID
+						jobIDCache[jobName] = prowJobID
+					}
+
+					scale := int64(windowDays)
+					rows = append(rows, models.ProwGARawTestDatum{
+						Release:    rel.Release,
+						WindowDays: windowDays,
+						TestID:     testID,
+						ProwJobID:  prowJobID,
+						SuiteID:    suiteID,
+						Passes:     int64(counts.success) * scale,
+						Failures:   int64(counts.total-counts.success-counts.flake) * scale,
+						Flakes:     int64(counts.flake) * scale,
+						Runs:       int64(counts.total) * scale,
+					})
+				}
+			}
+		}
+	}
+
+	if len(rows) > 0 {
+		if err := dbc.DB.CreateInBatches(rows, 500).Error; err != nil {
+			return fmt.Errorf("inserting GA raw test data: %w", err)
+		}
+	}
+
+	releasesWithRows := sets.New[string]()
+	for _, row := range rows {
+		releasesWithRows.Insert(row.Release)
+	}
+
+	for _, rel := range gaReleases {
+		if !releasesWithRows.Has(rel.Release) {
+			log.WithField("release", rel.Release).Warn("No GA seed data generated, skipping ga_data_loaded_date")
+			continue
+		}
+		gaDate := civil.DateOf(rel.GADate.UTC())
+		if err := dbc.DB.Model(&models.ReleaseDefinition{}).
+			Where("release = ?", rel.Release).
+			Update("ga_data_loaded_date", gaDate).Error; err != nil {
+			return fmt.Errorf("updating ga_data_loaded_date for %s: %w", rel.Release, err)
+		}
+	}
+
+	log.WithField("rows", len(rows)).
+		WithField("releases", len(gaReleases)).
+		Info("Seeded GA raw test data")
+	return nil
+}

pkg/api/componentreadiness/utils/utils.go

@@ -11,6 +11,7 @@ import (
 	"sync"
 	"time"
 
+	"cloud.google.com/go/civil"
 	"github.com/sirupsen/logrus"
 	"k8s.io/apimachinery/pkg/util/sets"
 
@@ -340,6 +341,22 @@ func GenerateTestDetailsURL(
 	return u.String(), nil
 }
 
+// GAWindows lists the distinct lookback periods (in days) for which GA base
+// data is fetched from BigQuery and stored in prow_ga_raw_test_data.
+var GAWindows = []int{1, 30, 90}
+
+// GAWindowStart returns the first day of a GA base window with the given
+// lookback period.
+func GAWindowStart(gaDate civil.Date, windowDays int) civil.Date {
+	return gaDate.AddDays(-windowDays)
+}
+
+// GAWindowEnd returns the exclusive end of the GA base window (the day after
+// the GA date).
+func GAWindowEnd(gaDate civil.Date) civil.Date {
+	return gaDate.AddDays(1)
+}
+
 // EnqueueAsync is a helper function to asynchronously enqueue something (like errors) on a channel from a synchronous context.
 func EnqueueAsync[T any](wg *sync.WaitGroup, channel chan T, things ...T) {
 	wg.Add(1)

pkg/bigquery/bqlabel/labels.go

@@ -100,6 +100,7 @@ const (
 	JobVariants                         QueryValue = "job-variants"
 	PRTestResults                       QueryValue = "pr-test-results"
 	CacheLookup                         QueryValue = "cache-lookup"
+	GATestStatusLoader                  QueryValue = "ga-test-status-loader"
 )
 
 // sanitizeLabelValue sanitizes a label value to meet BigQuery requirements: