Main

cmd/subresource-apiserver/main.go

@@ -165,7 +165,9 @@ func (h *handler) post(req *restful.Request, resp *restful.Response) {
 			})
 		},
 	)
-	if _, err := u.Upload(ctx, ns, sr, req.Request.Body); err != nil {
+	// Retrying 13 times should take around 2.73 minutes.
+	const maxRetriesForGetSourcePackage = 13
+	if _, err := u.Upload(ctx, ns, sr, maxRetriesForGetSourcePackage, req.Request.Body); err != nil {
 		h.logger.Warnf("failed to save image %s/%s: %v", ns, sr, err)
 		resp.WriteErrorString(http.StatusInternalServerError, "failed to save image")
 		return

pkg/sourceimage/uploader.go

@@ -25,6 +25,7 @@ import (
 	"io/ioutil"
 	"os"
 	"path"
+	"time"
 
 	"github.com/google/go-containerregistry/pkg/name"
 	"github.com/google/kf/v2/pkg/apis/kf/v1alpha1"
@@ -62,22 +63,39 @@ func (u *Uploader) Upload(
 	ctx context.Context,
 	spaceName string,
 	sourcePackageName string,
+	maxRetriesForGetSourcePackage int,
 	r io.Reader,
 ) (name.Reference, error) {
+	logger := logging.FromContext(ctx)
+
 	// Fetch the Space to read the configured container registry.
 	space, err := u.spaceLister.Get(spaceName)
 	if err != nil {
 		return nil, fmt.Errorf("failed to find Space: %v", err)
 	}
 
-	// Fetch the SourcePackage to lookup the spec.
-	sourcePackage, err := u.
-		sourcePackageLister.
-		SourcePackages(spaceName).
-		Get(sourcePackageName)
-	if err != nil {
-		return nil, fmt.Errorf("failed to find SourcePackage: %v", err)
+
+	// Fetch the SourcePackage to lookup the spec, retry on error with exponential backoff.
+	var sleepDuration = 10 * time.Millisecond
+	var sourcePackage *v1alpha1.SourcePackage
+	for i := 0; i <= maxRetriesForGetSourcePackage; i++ {
+		sourcePackage, err = u.
+			sourcePackageLister.
+			SourcePackages(spaceName).
+			Get(sourcePackageName)
+		if err == nil {
+			break
+		} else {
+			if (i == maxRetriesForGetSourcePackage) {
+				return nil, fmt.Errorf("failed to find SourcePackage, retries exhausted: %v", err)
+			} else {
+				logger.Warnf("failed to find SourcePackage: %s, retrying %dth time. Error: %v", sourcePackageName, i+1, err)
+				time.Sleep(sleepDuration)
+    			sleepDuration *= 2
+			}
+		}
 	}
+
 
 	// Ensure the sourcePackage is still pending. Otherwise, it is considered
 	// immutable.

pkg/sourceimage/uploader_test.go

@@ -85,6 +85,7 @@ func TestUploader(t *testing.T) {
 		name              string
 		spaceName         string
 		sourcePackageName string
+		maxRetriesForGetSourcePackage int
 		data              io.Reader
 		setup             func(t *testing.T, f fakes)
 		assert            func(t *testing.T, o output)
@@ -96,6 +97,7 @@ func TestUploader(t *testing.T) {
 			name:              "pushes the correct image",
 			spaceName:         "some-space",
 			sourcePackageName: "some-name",
+			maxRetriesForGetSourcePackage: 0,
 			setup: func(t *testing.T, f fakes) {
 				setupNormal(f, "some-space", "some-name")
 			},
@@ -128,6 +130,7 @@ func TestUploader(t *testing.T) {
 		{
 			name:      "getting Space fails",
 			spaceName: "some-space",
+			maxRetriesForGetSourcePackage: 0,
 			setup: func(t *testing.T, f fakes) {
 				f.fs.EXPECT().
 					Get("some-space").
@@ -141,25 +144,27 @@ func TestUploader(t *testing.T) {
 			name:              "getting SourcePackage fails",
 			spaceName:         "some-space",
 			sourcePackageName: "some-name",
+			maxRetriesForGetSourcePackage: 4,
 			setup: func(t *testing.T, f fakes) {
 				f.fs.EXPECT().Get("some-space")
 
 				f.fp.EXPECT().
 					SourcePackages("some-space").
-					Return(f.fnp)
+					Return(f.fnp).Times(5) // One initial attempt and then 4 retries.
 
 				f.fnp.EXPECT().
 					Get("some-name").
-					Return(nil, errors.New("some-error"))
+					Return(nil, errors.New("some-error")).Times(5) // One initial attempt and then 4 retries.
 			},
 			assert: func(t *testing.T, o output) {
-				testutil.AssertErrorsEqual(t, errors.New("failed to find SourcePackage: some-error"), o.err)
+				testutil.AssertErrorsEqual(t, errors.New("failed to find SourcePackage, retries exhausted: some-error"), o.err)
 			},
 		},
 		{
 			name:              "SourcePackage is not pending",
 			spaceName:         "some-space",
 			sourcePackageName: "some-name",
+			maxRetriesForGetSourcePackage: 0,
 			setup: func(t *testing.T, f fakes) {
 				f.fs.EXPECT().Get("some-space")
 
@@ -194,6 +199,7 @@ func TestUploader(t *testing.T) {
 			name:              "building and pushing image fails",
 			spaceName:         "some-space",
 			sourcePackageName: "some-name",
+			maxRetriesForGetSourcePackage: 0,
 			data:              strings.NewReader(normalData),
 			setup: func(t *testing.T, f fakes) {
 				setupNormal(f, "some-space", "some-name")
@@ -209,6 +215,7 @@ func TestUploader(t *testing.T) {
 			name:              "updating status fails",
 			spaceName:         "some-space",
 			sourcePackageName: "some-name",
+			maxRetriesForGetSourcePackage: 0,
 			data:              strings.NewReader(normalData),
 			setup: func(t *testing.T, f fakes) {
 				setupNormal(f, "some-space", "some-name")
@@ -224,6 +231,7 @@ func TestUploader(t *testing.T) {
 			name:              "saving data fails",
 			spaceName:         "some-space",
 			sourcePackageName: "some-name",
+			maxRetriesForGetSourcePackage: 0,
 			data:              &errReader{err: errors.New("some-error")},
 			setup: func(t *testing.T, f fakes) {
 				setupNormal(f, "some-space", "some-name")
@@ -236,6 +244,7 @@ func TestUploader(t *testing.T) {
 			name:              "checksum doesn't match",
 			spaceName:         "some-space",
 			sourcePackageName: "some-name",
+			maxRetriesForGetSourcePackage: 0,
 
 			// NOTE: the variable normalData is associated with
 			// normalChecksumValue which is set by the setupNormal function.
@@ -251,6 +260,7 @@ func TestUploader(t *testing.T) {
 			name:              "size doesn't match spec",
 			spaceName:         "some-space",
 			sourcePackageName: "some-name",
+			maxRetriesForGetSourcePackage: 0,
 			data:              strings.NewReader(normalData),
 			setup: func(t *testing.T, f fakes) {
 				setupWithSize(f, "some-space", "some-name", uint64(len(normalData)+1))
@@ -263,6 +273,7 @@ func TestUploader(t *testing.T) {
 			name:              "unknown checksum type",
 			spaceName:         "some-space",
 			sourcePackageName: "some-name",
+			maxRetriesForGetSourcePackage: 0,
 			data:              strings.NewReader(normalData),
 			setup: func(t *testing.T, f fakes) {
 				setup(f, "some-space", "some-name", "invalid", uint64(len(normalData)))
@@ -275,6 +286,7 @@ func TestUploader(t *testing.T) {
 			name:              "limits how much it reads",
 			spaceName:         "some-space",
 			sourcePackageName: "some-name",
+			maxRetriesForGetSourcePackage: 0,
 			// This reader will never stop feeding the test data. Unless it
 			// has properly guarded itself by limiting how much data it reads,
 			// it will timeout.
@@ -336,6 +348,7 @@ func TestUploader(t *testing.T) {
 				context.Background(),
 				tc.spaceName,
 				tc.sourcePackageName,
+				tc.maxRetriesForGetSourcePackage,
 				tc.data,
 			)