Merge pull request #1268 from michelle192837/retry

Add retry when snapshotting configuration.

Author: google-oss-prow[bot]

config/cache_test.go

@@ -100,14 +100,16 @@ func Test_ReadGCS(t *testing.T) {
 		t.Run(test.name, func(t *testing.T) {
 			cache = test.currentCache
 			client := fake.Client{
-				Opener: fake.Opener{},
+				Opener: fake.Opener{
+					Paths: map[gcs.Path]fake.Object{},
+				},
 			}
 			expectedAttrs := &storage.ReaderObjectAttrs{
 				LastModified: test.remoteLastModified,
 				Generation:   test.remoteGeneration,
 			}
 
-			client.Opener[mustPath("gs://example")] = fake.Object{
+			client.Opener.Paths[mustPath("gs://example")] = fake.Object{
 				Data:  string(test.remoteData),
 				Attrs: expectedAttrs,
 			}

config/snapshot/BUILD.bazel

@@ -9,6 +9,7 @@ go_library(
         "//config:go_default_library",
         "//pb/config:go_default_library",
         "//util/gcs:go_default_library",
+        "@com_github_sethvargo_go_retry//:go_default_library",
         "@com_github_sirupsen_logrus//:go_default_library",
         "@com_google_cloud_go_storage//:go_default_library",
     ],

config/snapshot/config_snapshot.go

@@ -25,6 +25,7 @@ import (
 	"github.com/GoogleCloudPlatform/testgrid/config"
 	configpb "github.com/GoogleCloudPlatform/testgrid/pb/config"
 	"github.com/GoogleCloudPlatform/testgrid/util/gcs"
+	"github.com/sethvargo/go-retry"
 	"github.com/sirupsen/logrus"
 )
 
@@ -119,6 +120,22 @@ func updateHash(ctx context.Context, client gcs.Opener, configPath gcs.Path) (*C
 }
 
 func fetchConfig(ctx context.Context, client gcs.Opener, configPath gcs.Path) (*configpb.Configuration, *storage.ReaderObjectAttrs, error) {
+	backoff := retry.WithMaxRetries(2, retry.NewExponential(5*time.Second))
+
+	var cfg *configpb.Configuration
+	var attrs *storage.ReaderObjectAttrs
+	err := retry.Do(ctx, backoff, func(innerCtx context.Context) error {
+		var onceErr error
+		cfg, attrs, onceErr = fetchConfigOnce(innerCtx, client, configPath)
+		if onceErr != nil {
+			return retry.RetryableError(onceErr)
+		}
+		return nil
+	})
+	return cfg, attrs, err
+}
+
+func fetchConfigOnce(ctx context.Context, client gcs.Opener, configPath gcs.Path) (*configpb.Configuration, *storage.ReaderObjectAttrs, error) {
 	r, attrs, err := client.Open(ctx, configPath)
 	if err != nil {
 		return nil, nil, fmt.Errorf("open: %w", err)

config/snapshot/config_snapshot_test.go

@@ -73,7 +73,7 @@ func TestObserve_OnInit(t *testing.T) {
 			defer cancel()
 
 			client := fakeClient()
-			client.Opener[*path] = fake.Object{
+			client.Opener.Paths[*path] = fake.Object{
 				Data: string(mustMarshalConfig(test.config)),
 				Attrs: &storage.ReaderObjectAttrs{
 					Generation: test.configGeneration,
@@ -108,6 +108,87 @@ func TestObserve_OnInit(t *testing.T) {
 	}
 }
 
+func TestObserve_OnInitRetry(t *testing.T) {
+	tests := []struct {
+		name             string
+		config           *configpb.Configuration
+		configGeneration int64
+		openErr          error
+		openOnRetry      bool
+		expectInitial    *configpb.Dashboard
+		expectError      bool
+	}{
+		{
+			name: "Reads config on retry",
+			config: &configpb.Configuration{
+				Dashboards: []*configpb.Dashboard{
+					{
+						Name: "dashboard",
+					},
+				},
+			},
+			openErr:     errors.New("fake error"),
+			openOnRetry: true,
+			expectInitial: &configpb.Dashboard{
+				Name: "dashboard",
+			},
+		},
+		{
+			name:        "Returns error if config isn't present on retry",
+			openErr:     errors.New("fake error"),
+			expectError: true,
+		},
+	}
+
+	path, err := gcs.NewPath("gs://config/example")
+	if err != nil {
+		t.Fatal("could not path")
+	}
+
+	for _, test := range tests {
+		t.Run(test.name, func(t *testing.T) {
+			ctx, cancel := context.WithCancel(context.Background())
+			defer cancel()
+
+			client := fakeClient()
+			client.Opener.Paths[*path] = fake.Object{
+				Data: string(mustMarshalConfig(test.config)),
+				Attrs: &storage.ReaderObjectAttrs{
+					Generation: 1,
+				},
+				OpenErr:     test.openErr,
+				OpenOnRetry: test.openOnRetry,
+			}
+			client.Stater[*path] = fake.Stat{
+				Attrs: storage.ObjectAttrs{
+					Generation: 1,
+				},
+			}
+
+			snaps, err := Observe(ctx, nil, client, *path, nil)
+
+			if !test.expectError && err != nil {
+				t.Errorf("Observe() got unexpected error: %v", err)
+			} else if test.expectError && err == nil {
+				t.Errorf("Observe() did not error as expected.")
+			}
+
+			if test.expectInitial == nil {
+				return
+			}
+
+			select {
+			case cs := <-snaps:
+				if result := cs.Dashboards["dashboard"]; !proto.Equal(result, test.expectInitial) {
+					t.Errorf("got dashboard %v, expected %v", result, test.expectInitial)
+				}
+			case <-time.After(30 * time.Second):
+				t.Error("expected an initial snapshot, but got none")
+			}
+		})
+	}
+}
+
 func TestObserve_OnTick(t *testing.T) {
 	tests := []struct {
 		name             string
@@ -169,7 +250,7 @@ func TestObserve_OnTick(t *testing.T) {
 			defer cancel()
 
 			client := fakeClient()
-			client.Opener[*path] = fake.Object{
+			client.Opener.Paths[*path] = fake.Object{
 				Data: string(mustMarshalConfig(initialConfig)),
 				Attrs: &storage.ReaderObjectAttrs{
 					Generation: 1,
@@ -190,7 +271,7 @@ func TestObserve_OnTick(t *testing.T) {
 			<-snaps
 
 			// Change the config
-			client.Opener[*path] = fake.Object{
+			client.Opener.Paths[*path] = fake.Object{
 				Data: string(mustMarshalConfig(test.config)),
 				Attrs: &storage.ReaderObjectAttrs{
 					Generation: test.configGeneration,
@@ -350,7 +431,7 @@ func TestObserve_Data(t *testing.T) {
 			defer cancel()
 
 			client := fakeClient()
-			client.Opener[*path] = fake.Object{
+			client.Opener.Paths[*path] = fake.Object{
 				Data: string(mustMarshalConfig(test.config)),
 				Attrs: &storage.ReaderObjectAttrs{
 					Generation: 1,
@@ -394,7 +475,10 @@ func fakeClient() *fake.ConditionalClient {
 			Uploader: fake.Uploader{},
 			Client: fake.Client{
 				Lister: fake.Lister{},
-				Opener: fake.Opener{},
+				Opener: fake.Opener{
+					Paths: map[gcs.Path]fake.Object{},
+					Lock:  &sync.RWMutex{},
+				},
 			},
 			Stater: fake.Stater{},
 		},

go.mod

@@ -17,6 +17,7 @@ require (
 	github.com/hashicorp/go-multierror v1.1.1
 	github.com/prometheus/client_golang v1.11.1
 	github.com/prometheus/client_model v0.3.0
+	github.com/sethvargo/go-retry v0.2.4
 	github.com/sirupsen/logrus v1.9.3
 	google.golang.org/api v0.134.0
 	google.golang.org/genproto v0.0.0-20230731193218-e0aa005b6bdf

go.sum

@@ -1152,6 +1152,8 @@ github.com/rogpeppe/go-internal v1.11.0 h1:cWPaGQEPrBb5/AsnsZesgZZ9yb1OQ+GOISoDN
 github.com/rogpeppe/go-internal v1.11.0/go.mod h1:ddIwULY96R17DhadqLgMfk9H9tvdUzkipdSkR5nkCZA=
 github.com/ruudk/golang-pdf417 v0.0.0-20181029194003-1af4ab5afa58/go.mod h1:6lfFZQK844Gfx8o5WFuvpxWRwnSoipWe/p622j1v06w=
 github.com/ruudk/golang-pdf417 v0.0.0-20201230142125-a7e3863a1245/go.mod h1:pQAZKsJ8yyVxGRWYNEm9oFB8ieLgKFnamEyDmSA0BRk=
+github.com/sethvargo/go-retry v0.2.4 h1:T+jHEQy/zKJf5s95UkguisicE0zuF9y7+/vgz08Ocec=
+github.com/sethvargo/go-retry v0.2.4/go.mod h1:1afjQuvh7s4gflMObvjLPaWgluLLyhA1wmVZ6KLpICw=
 github.com/sirupsen/logrus v1.2.0/go.mod h1:LxeOpSwHxABJmUn/MG1IvRgCAasNZTLOkJPxbbu5VWo=
 github.com/sirupsen/logrus v1.4.2/go.mod h1:tLMulIdttU9McNUspp0xgXVQah82FyeX6MwdIuYE2rE=
 github.com/sirupsen/logrus v1.6.0/go.mod h1:7uNnSEd1DgxDLC74fIahvMZmmYsHGZGEOFrfsX/uA88=

pkg/updater/persist_test.go

@@ -223,7 +223,9 @@ func TestFixPersistent(t *testing.T) {
 				Uploader: fake.Uploader{},
 				Client: fake.Client{
 					Opener: fake.Opener{
-						*path: tc.currently,
+						Paths: map[gcs.Path]fake.Object{
+							*path: tc.currently,
+						},
 					},
 				},
 			}

pkg/updater/read_test.go

@@ -1178,7 +1178,10 @@ func TestReadColumns(t *testing.T) {
 			defer cancel()
 			client := fakeClient{
 				Lister: fake.Lister{},
-				Opener: fake.Opener{},
+				Opener: fake.Opener{
+					Paths: map[gcs.Path]fake.Object{},
+					Lock:  &sync.RWMutex{},
+				},
 			}
 
 			builds := addBuilds(&client, path, tc.builds...)
@@ -1668,7 +1671,10 @@ func TestReadResult(t *testing.T) {
 			defer cancel()
 			client := fakeClient{
 				Lister: fake.Lister{},
-				Opener: fake.Opener{},
+				Opener: fake.Opener{
+					Paths: map[gcs.Path]fake.Object{},
+					Lock:  &sync.RWMutex{},
+				},
 			}
 
 			fi := fakeIterator{}
@@ -1680,7 +1686,7 @@ func TestReadResult(t *testing.T) {
 				fi.Objects = append(fi.Objects, storage.ObjectAttrs{
 					Name: p.Object(),
 				})
-				client.Opener[*p] = fo
+				client.Opener.Paths[*p] = fo
 			}
 			client.Lister[path] = fi
 
@@ -1850,7 +1856,9 @@ func TestReadSuites(t *testing.T) {
 			defer cancel()
 			client := fakeClient{
 				Lister: fake.Lister{},
-				Opener: fake.Opener{},
+				Opener: fake.Opener{
+					Paths: map[gcs.Path]fake.Object{},
+				},
 			}
 
 			fi := fakeIterator{
@@ -1864,7 +1872,7 @@ func TestReadSuites(t *testing.T) {
 				fi.Objects = append(fi.Objects, storage.ObjectAttrs{
 					Name: p.Object(),
 				})
-				client.Opener[*p] = fo
+				client.Opener.Paths[*p] = fo
 			}
 			client.Lister[path] = fi
 
@@ -1898,6 +1906,10 @@ func TestReadSuites(t *testing.T) {
 }
 
 func addBuilds(fc *fake.Client, path gcs.Path, s ...fakeBuild) []gcs.Build {
+	if fc.Opener.Lock != nil {
+		fc.Opener.Lock.Lock()
+		defer fc.Opener.Lock.Unlock()
+	}
 	var builds []gcs.Build
 	for _, build := range s {
 		buildPath := resolveOrDie(&path, build.id+"/")
@@ -1909,25 +1921,25 @@ func addBuilds(fc *fake.Client, path gcs.Path, s ...fakeBuild) []gcs.Build {
 			fi.Objects = append(fi.Objects, storage.ObjectAttrs{
 				Name: p.Object(),
 			})
-			fc.Opener[*p] = *build.podInfo
+			fc.Opener.Paths[*p] = *build.podInfo
 		}
 		if build.started != nil {
 			p := resolveOrDie(buildPath, "started.json")
 			fi.Objects = append(fi.Objects, storage.ObjectAttrs{
 				Name: p.Object(),
 			})
-			fc.Opener[*p] = *build.started
+			fc.Opener.Paths[*p] = *build.started
 		}
 		if build.finished != nil {
 			p := resolveOrDie(buildPath, "finished.json")
 			fi.Objects = append(fi.Objects, storage.ObjectAttrs{
 				Name: p.Object(),
 			})
-			fc.Opener[*p] = *build.finished
+			fc.Opener.Paths[*p] = *build.finished
 		}
 		if len(build.passed)+len(build.failed) > 0 {
 			p := resolveOrDie(buildPath, "junit_automatic.xml")
-			fc.Opener[*p] = fake.Object{Data: makeJunit(build.passed, build.failed)}
+			fc.Opener.Paths[*p] = fake.Object{Data: makeJunit(build.passed, build.failed)}
 			fi.Objects = append(fi.Objects, storage.ObjectAttrs{
 				Name: p.Object(),
 			})
@@ -1937,7 +1949,7 @@ func addBuilds(fc *fake.Client, path gcs.Path, s ...fakeBuild) []gcs.Build {
 			fi.Objects = append(fi.Objects, storage.ObjectAttrs{
 				Name: p.Object(),
 			})
-			fc.Opener[*p] = fo
+			fc.Opener.Paths[*p] = fo
 		}
 		fc.Lister[*buildPath] = fi
 	}

pkg/updater/updater_test.go

@@ -389,13 +389,15 @@ func TestUpdate(t *testing.T) {
 					Uploader: fakeUploader{},
 					Client: fakeClient{
 						Lister: fakeLister{},
-						Opener: fakeOpener{},
+						Opener: fakeOpener{
+							Paths: map[gcs.Path]fake.Object{},
+						},
 					},
 				},
 				Lock: &sync.RWMutex{},
 			}
 
-			client.Opener[configPath] = fakeObject{
+			client.Opener.Paths[configPath] = fakeObject{
 				Data: func() string {
 					b, err := config.MarshalBytes(tc.config)
 					if err != nil {
@@ -2028,13 +2030,15 @@ func TestInflateDropAppend(t *testing.T) {
 					Uploader: fakeUploader{},
 					Client: fakeClient{
 						Lister: fakeLister{},
-						Opener: fakeOpener{},
+						Opener: fakeOpener{
+							Paths: map[gcs.Path]fake.Object{},
+						},
 					},
 				},
 			}
 
 			if tc.current != nil {
-				client.Opener[uploadPath] = *tc.current
+				client.Opener.Paths[uploadPath] = *tc.current
 			}
 
 			buildsPath := newPathOrDie("gs://" + tc.group.GcsPrefix)
@@ -2079,13 +2083,15 @@ func TestInflateDropAppend(t *testing.T) {
 				}
 				t.Logf("InflateDropAppend() generated a binary diff (-want +got):\n%s", diff)
 				fakeDownloader := fakeOpener{
-					uploadPath: {Data: string(actual[uploadPath].Buf)},
+					Paths: map[gcs.Path]fake.Object{
+						uploadPath: {Data: string(actual[uploadPath].Buf)},
+					},
 				}
 				actualGrid, _, err := gcs.DownloadGrid(ctx, fakeDownloader, uploadPath)
 				if err != nil {
 					t.Errorf("actual gcs.DownloadGrid() got unexpected error: %v", err)
 				}
-				fakeDownloader[uploadPath] = fakeObject{Data: string(tc.expected.Buf)}
+				fakeDownloader.Paths[uploadPath] = fakeObject{Data: string(tc.expected.Buf)}
 				expectedGrid, _, err := gcs.DownloadGrid(ctx, fakeDownloader, uploadPath)
 				if err != nil {
 					t.Errorf("expected gcs.DownloadGrid() got unexpected error: %v", err)

repos.bzl

@@ -1227,6 +1227,14 @@ def go_repositories():
         sum = "h1:K1Xf3bKttbF+koVGaX5xngRIZ5bVjbmPnaxE/dR08uY=",
         version = "v0.0.0-20201230142125-a7e3863a1245",
     )
+    go_repository(
+        name = "com_github_sethvargo_go_retry",
+        build_file_generation = "on",
+        build_file_proto_mode = "disable",
+        importpath = "github.com/sethvargo/go-retry",
+        sum = "h1:T+jHEQy/zKJf5s95UkguisicE0zuF9y7+/vgz08Ocec=",
+        version = "v0.2.4",
+    )
 
     go_repository(
         name = "com_github_sirupsen_logrus",