Optimize prow loader with bulk writer, COPY protocol, and GCS improvements

pkg/api/jobrunevents/job_run_events.go

@@ -5,7 +5,6 @@ import (
 	"encoding/json"
 	"errors"
 	"fmt"
-	"regexp"
 	"strings"
 
 	"cloud.google.com/go/storage"
@@ -16,9 +15,6 @@ import (
 	"github.com/openshift/sippy/pkg/db"
 )
 
-// eventsJSONRegex matches paths like artifacts/*e2e*/gather-extra/artifacts/events.json
-var eventsJSONRegex = regexp.MustCompile(`gather-extra/artifacts/events\.json$`)
-
 // KubeEvent represents a flattened Kubernetes Event for the API response
 type KubeEvent struct {
 	FirstTimestamp string `json:"firstTimestamp"`
@@ -76,17 +72,17 @@ func JobRunEvents(gcsClient *storage.Client, dbc *db.DB, jobRunID int64, gcsBuck
 	}
 
 	gcsJobRun := gcs.NewGCSJobRun(gcsClient.Bucket(gcsBucket), gcsPath)
-	matches, err := gcsJobRun.FindAllMatches([]*regexp.Regexp{eventsJSONRegex})
+	matches, err := gcsJobRun.FindAllMatches(context.TODO(), gcs.GlobEventsJSON)
 	if err != nil {
 		return &EventListResponse{JobRunURL: jobRunURL}, err
 	}
 
-	if len(matches) == 0 || len(matches[0]) == 0 {
+	if len(matches) == 0 {
 		logger.Info("no events.json file found")
 		return &EventListResponse{Items: []KubeEvent{}, JobRunURL: jobRunURL}, nil
 	}
 
-	eventsPath := matches[0][0]
+	eventsPath := matches[0]
 	logger.WithField("events_path", eventsPath).Info("found events.json")
 
 	content, err := gcsJobRun.GetContent(context.TODO(), eventsPath)

pkg/api/jobrunintervals/job_run_intervals.go

@@ -5,7 +5,6 @@ import (
 	"encoding/json"
 	"errors"
 	"fmt"
-	"regexp"
 	"strings"
 
 	"cloud.google.com/go/storage"
@@ -45,10 +44,15 @@ func JobRunIntervals(gcsClient *storage.Client, dbc *db.DB, jobRunID int64, gcsB
 
 	gcsJobRun := gcs.NewGCSJobRun(gcsClient.Bucket(gcsBucket), gcsPath)
 	intervals := &apitype.EventIntervalList{}
-	intervalFiles, err := gcsJobRun.FindAllMatches([]*regexp.Regexp{gcs.GetIntervalFile()})
+	intervalFiles, err := gcsJobRun.FindAllMatches(context.TODO(), gcs.GlobIntervalsJSON)
 	if err != nil {
 		return intervals, err
 	}
+	timelineFiles, err := gcsJobRun.FindAllMatches(context.TODO(), gcs.GlobTimelinesJSON)
+	if err != nil {
+		return intervals, err
+	}
+	intervalFiles = append(intervalFiles, timelineFiles...)
 
 	// We will often match multiple files here, one for upgrade phase, one for conformance
 	// testing phase. For now, we return them all, and each interval has a filename it
@@ -57,7 +61,7 @@ func JobRunIntervals(gcsClient *storage.Client, dbc *db.DB, jobRunID int64, gcsB
 		logger.Info("no interval files found")
 		return intervals, nil
 	}
-	logger.WithField("interval_files", intervalFiles[0]).Info("found interval files")
+	logger.WithField("interval_files", intervalFiles).Info("found interval files")
 
 	// Now that we have the list of all matching interval files, if the user specified one, return only
 	// intervals from that file. If they didn't, make a best guess on the default to load, our minimal
@@ -66,7 +70,7 @@ func JobRunIntervals(gcsClient *storage.Client, dbc *db.DB, jobRunID int64, gcsB
 	intervalFilesAvailable := []string{}
 
 	// Find the full path to the filename specified:
-	for _, fp := range intervalFiles[0] {
+	for _, fp := range intervalFiles {
 		// Get the base filename we'll add to each incoming interval:
 		tokens := strings.Split(fp, "/")
 		baseFile := tokens[len(tokens)-1]

pkg/dataloader/prowloader/bigqueryjobs.go

@@ -64,6 +64,7 @@ func (pl *ProwLoader) fetchProwJobsFromOpenShiftBigQuery() ([]prow.ProwJob, []er
 		"FROM `ci_analysis_us.jobs` "+
 		`WHERE TIMESTAMP(prowjob_completion) > @queryFrom
 	       AND prowjob_url IS NOT NULL
+	       AND prowjob_state NOT IN ('pending', 'triggered')
 	       ORDER BY prowjob_start_ts`)
 	query.Parameters = []bigquery.QueryParameter{
 		{

pkg/dataloader/prowloader/extract_test_cases_test.go

@@ -0,0 +1,155 @@
+package prowloader
+
+import (
+	"testing"
+
+	"github.com/stretchr/testify/assert"
+
+	"github.com/openshift/sippy/pkg/apis/junit"
+	sippyprocessingv1 "github.com/openshift/sippy/pkg/apis/sippyprocessing/v1"
+)
+
+func TestExtractTestCases(t *testing.T) {
+	failMsg := "something went wrong"
+
+	tests := []struct {
+		name     string
+		suite    *junit.TestSuite
+		expected map[testCaseKey]*testCaseEntry
+	}{
+		{
+			name: "passing test",
+			suite: &junit.TestSuite{
+				Name: "openshift-tests",
+				TestCases: []*junit.TestCase{
+					{Name: "test-a", Duration: 1.5},
+				},
+			},
+			expected: map[testCaseKey]*testCaseEntry{
+				{SuiteName: "openshift-tests", TestName: "test-a"}: {
+					TestName:  "test-a",
+					SuiteName: "openshift-tests",
+					Status:    int(sippyprocessingv1.TestStatusSuccess),
+					Duration:  1.5,
+				},
+			},
+		},
+		{
+			name: "failing test with output",
+			suite: &junit.TestSuite{
+				Name: "openshift-tests",
+				TestCases: []*junit.TestCase{
+					{Name: "test-a", Duration: 2.0, FailureOutput: &junit.FailureOutput{Output: failMsg}},
+				},
+			},
+			expected: map[testCaseKey]*testCaseEntry{
+				{SuiteName: "openshift-tests", TestName: "test-a"}: {
+					TestName:  "test-a",
+					SuiteName: "openshift-tests",
+					Status:    int(sippyprocessingv1.TestStatusFailure),
+					Duration:  2.0,
+					Output:    &failMsg,
+				},
+			},
+		},
+		{
+			name: "skipped test excluded",
+			suite: &junit.TestSuite{
+				Name: "openshift-tests",
+				TestCases: []*junit.TestCase{
+					{Name: "test-a", SkipMessage: &junit.SkipMessage{Message: "skipped"}},
+				},
+			},
+			expected: map[testCaseKey]*testCaseEntry{},
+		},
+		{
+			name: "flake from pass then fail",
+			suite: &junit.TestSuite{
+				Name: "openshift-tests",
+				TestCases: []*junit.TestCase{
+					{Name: "test-a", Duration: 1.0},
+					{Name: "test-a", Duration: 2.0, FailureOutput: &junit.FailureOutput{Output: failMsg}},
+				},
+			},
+			expected: map[testCaseKey]*testCaseEntry{
+				{SuiteName: "openshift-tests", TestName: "test-a"}: {
+					TestName:  "test-a",
+					SuiteName: "openshift-tests",
+					Status:    int(sippyprocessingv1.TestStatusFlake),
+					Duration:  1.0,
+					Output:    &failMsg,
+				},
+			},
+		},
+		{
+			name: "dotted names do not collide",
+			suite: &junit.TestSuite{
+				Name: "a.b",
+				TestCases: []*junit.TestCase{
+					{Name: "c", Duration: 1.0},
+				},
+				Children: []*junit.TestSuite{
+					{
+						Name: "a",
+						TestCases: []*junit.TestCase{
+							{Name: "b.c", Duration: 2.0},
+						},
+					},
+				},
+			},
+			expected: map[testCaseKey]*testCaseEntry{
+				{SuiteName: "a.b", TestName: "c"}: {
+					TestName:  "c",
+					SuiteName: "a.b",
+					Status:    int(sippyprocessingv1.TestStatusSuccess),
+					Duration:  1.0,
+				},
+				{SuiteName: "a", TestName: "b.c"}: {
+					TestName:  "b.c",
+					SuiteName: "a",
+					Status:    int(sippyprocessingv1.TestStatusSuccess),
+					Duration:  2.0,
+				},
+			},
+		},
+		{
+			name: "child suite uses its own name",
+			suite: &junit.TestSuite{
+				Name: "openshift-tests",
+				TestCases: []*junit.TestCase{
+					{Name: "parent-test", Duration: 1.0},
+				},
+				Children: []*junit.TestSuite{
+					{
+						Name: "k8s.io",
+						TestCases: []*junit.TestCase{
+							{Name: "child-test", Duration: 3.0},
+						},
+					},
+				},
+			},
+			expected: map[testCaseKey]*testCaseEntry{
+				{SuiteName: "openshift-tests", TestName: "parent-test"}: {
+					TestName:  "parent-test",
+					SuiteName: "openshift-tests",
+					Status:    int(sippyprocessingv1.TestStatusSuccess),
+					Duration:  1.0,
+				},
+				{SuiteName: "k8s.io", TestName: "child-test"}: {
+					TestName:  "child-test",
+					SuiteName: "k8s.io",
+					Status:    int(sippyprocessingv1.TestStatusSuccess),
+					Duration:  3.0,
+				},
+			},
+		},
+	}
+
+	for _, tt := range tests {
+		t.Run(tt.name, func(t *testing.T) {
+			testCases := make(map[testCaseKey]*testCaseEntry)
+			extractTestCases(tt.suite, testCases)
+			assert.Equal(t, tt.expected, testCases)
+		})
+	}
+}

pkg/dataloader/prowloader/gcs/gcs_jobrun.go

@@ -16,45 +16,24 @@ import (
 )
 
 const TestFailureSummaryFilePrefix = "risk-analysis"
-const ClusterDataFilePrefix = "cluster-data_"
-const JunitRegExStr = "\\/.*junit.*xml"
-const intervalFilesRegExStr = "\\/(e2e-events|e2e-timelines).*json"
-
-var (
-	defaultRiskAnalysisSummaryFileRegEx *regexp.Regexp
-	defaultClusterDataFileRegEx         *regexp.Regexp
-	defaultJunitFileRegEx               *regexp.Regexp
-	intervalFilesRegex                  *regexp.Regexp
+
+const (
+	GlobJunitXML      = "**junit*.xml"
+	GlobEventsJSON    = "**/gather-extra/artifacts/events.json"
+	GlobIntervalsJSON = "**e2e-events*.json"
+	GlobTimelinesJSON = "**e2e-timelines*.json"
+	GlobClusterData   = "**/cluster-data_*.json"
 )
 
+var defaultRiskAnalysisSummaryFileRegEx *regexp.Regexp
+
 func GetDefaultRiskAnalysisSummaryFile() *regexp.Regexp {
 	if defaultRiskAnalysisSummaryFileRegEx == nil {
 		defaultRiskAnalysisSummaryFileRegEx = regexp.MustCompile(fmt.Sprintf("%s.json", TestFailureSummaryFilePrefix))
 	}
 	return defaultRiskAnalysisSummaryFileRegEx
 }
 
-func GetDefaultClusterDataFile() *regexp.Regexp {
-	if defaultClusterDataFileRegEx == nil {
-		defaultClusterDataFileRegEx = regexp.MustCompile(fmt.Sprintf("%s.*json", ClusterDataFilePrefix))
-	}
-	return defaultClusterDataFileRegEx
-}
-
-func GetDefaultJunitFile() *regexp.Regexp {
-	if defaultJunitFileRegEx == nil {
-		defaultJunitFileRegEx = regexp.MustCompile(JunitRegExStr)
-	}
-	return defaultJunitFileRegEx
-}
-
-func GetIntervalFile() *regexp.Regexp {
-	if intervalFilesRegex == nil {
-		intervalFilesRegex = regexp.MustCompile(intervalFilesRegExStr)
-	}
-	return intervalFilesRegex
-}
-
 type GCSJobRun struct {
 	// retrieval mechanisms
 	bkt *storage.BucketHandle
@@ -76,24 +55,21 @@ func (j *GCSJobRun) SetGCSJunitPaths(paths []string) {
 	j.gcsJunitPaths = paths
 }
 
-func (j *GCSJobRun) GetGCSJunitPaths() ([]string, error) {
+func (j *GCSJobRun) GetGCSJunitPaths(ctx context.Context) ([]string, error) {
 	if len(j.gcsJunitPaths) == 0 {
-		matches, err := j.FindAllMatches([]*regexp.Regexp{GetDefaultJunitFile()})
+		matches, err := j.FindAllMatches(ctx, GlobJunitXML)
 		if err != nil {
 			return nil, err
 		}
-
-		if len(matches) > 0 {
-			j.gcsJunitPaths = matches[0]
-		}
+		j.gcsJunitPaths = matches
 	}
 
 	return j.gcsJunitPaths, nil
 }
 
 func (j *GCSJobRun) GetCombinedJUnitTestSuites(ctx context.Context) (*junit.TestSuites, error) {
 	testSuites := &junit.TestSuites{}
-	junitPaths, err := j.GetGCSJunitPaths()
+	junitPaths, err := j.GetGCSJunitPaths(ctx)
 	if err != nil {
 		return nil, err
 	}
@@ -135,19 +111,7 @@ func (j *GCSJobRun) GetContent(ctx context.Context, path string) ([]byte, error)
 		return content, nil
 	}
 
-	// Get an Object handle for the path
 	obj := j.bkt.Object(path)
-
-	// use the object attributes to try to get the latest generation to try to retrieve the data without getting a cached
-	// version of data that does not match the latest content.  I don't know if this will work, but in the easy case
-	// it doesn't seem to fail.
-	objAttrs, err := obj.Attrs(ctx)
-	if err != nil {
-		return nil, fmt.Errorf("error reading GCS attributes for jobrun: %w", err)
-	}
-	obj = obj.Generation(objAttrs.Generation)
-
-	// Get an io.Reader for the object.
 	gcsReader, err := obj.NewReader(ctx)
 	if err != nil {
 		return nil, fmt.Errorf("error reading GCS content for jobrun: %w", err)
@@ -196,20 +160,19 @@ func (j *GCSJobRun) FindFirstFile(root string, filename *regexp.Regexp) []byte {
 	return nil
 }
 
-// FindAllMatches takes an array of regexes
-// and compares the name of the object in gcs
-// with each regex for a match
-// each regex that matches will get the attribute name
-// in the returned matches with the index matching the regex
-func (j *GCSJobRun) FindAllMatches(filenames []*regexp.Regexp) ([][]string, error) {
-	if len(filenames) < 1 {
-		return nil, nil
+// FindAllMatches lists GCS objects under the job run path that match the
+// given glob pattern, using server-side filtering via MatchGlob.
+func (j *GCSJobRun) FindAllMatches(ctx context.Context, glob string) ([]string, error) {
+	q := &storage.Query{
+		Prefix:    j.gcsProwJobPath,
+		MatchGlob: glob,
+	}
+	if err := q.SetAttrSelection([]string{"Name"}); err != nil {
+		return nil, errors.Wrap(err, "error setting attribute selection")
 	}
-	matches := make([][]string, len(filenames))
 
-	it := j.bkt.Objects(context.Background(), &storage.Query{
-		Prefix: j.gcsProwJobPath,
-	})
+	var matches []string
+	it := j.bkt.Objects(ctx, q)
 	for {
 		attrs, err := it.Next()
 		if errors.Is(err, iterator.Done) {
@@ -218,15 +181,7 @@ func (j *GCSJobRun) FindAllMatches(filenames []*regexp.Regexp) ([][]string, erro
 		if err != nil {
 			return nil, errors.Wrap(err, "error reading GCS attributes for job run")
 		}
-
-		for i, filename := range filenames {
-			if matches[i] == nil {
-				matches[i] = make([]string, 0)
-			}
-			if filename.MatchString(attrs.Name) {
-				matches[i] = append(matches[i], attrs.Name)
-			}
-		}
+		matches = append(matches, attrs.Name)
 	}
 
 	return matches, nil