fix(pkg/gitreceive): move the object polling logic into storage

and use it in build.go. also add tests for the new polling logic

Author: Aaron Schlesinger

builder

@@ -20,7 +20,6 @@ import (
 
 	"k8s.io/kubernetes/pkg/api"
 	client "k8s.io/kubernetes/pkg/client/unversioned"
-	"k8s.io/kubernetes/pkg/util/wait"
 )
 
 // repoCmd returns exec.Command(first, others...) with its current working directory repoDir
@@ -224,15 +223,13 @@ func build(conf *Config, s3Client *storage.Client, kubeClient *client.Client, fs
 		slugBuilderInfo.PushKey(),
 	)
 	// poll the s3 server to ensure the slug exists
-	err = wait.PollImmediate(conf.ObjectStorageTickDuration(), conf.ObjectStorageWaitDuration(), func() (bool, error) {
-		exists, err := storage.ObjectExists(s3Client, conf.Bucket, slugBuilderInfo.PushKey())
-		if err != nil {
-			return false, fmt.Errorf("Checking if object %s/%s exists (%s)", conf.Bucket, slugBuilderInfo.PushKey(), err)
-		}
-		return exists, nil
-	})
-
-	if err != nil {
+	if err := storage.WaitForObject(
+		s3Client,
+		conf.Bucket,
+		slugBuilderInfo.PushKey(),
+		conf.ObjectStorageTickDuration(),
+		conf.ObjectStorageWaitDuration(),
+	); err != nil {
 		return fmt.Errorf("Timed out waiting for object in storage, aborting build (%s)", err)
 	}
 	log.Info("Build complete.")

pkg/gitreceive/build.go

@@ -1,7 +1,10 @@
 package storage
 
 import (
+	"errors"
+	"fmt"
 	"io"
+	"time"
 
 	s3 "github.com/minio/minio-go"
 )
@@ -18,21 +21,54 @@ const (
 // - false, err with the appropriate error if the statter failed
 // - true, nil if the statter succeeded and reported the object exists
 func ObjectExists(statter ObjectStatter, bucketName, objKey string) (bool, error) {
-	objInfo, err := statter.StatObject(bucketName, objKey)
+	_, err := statter.StatObject(bucketName, objKey)
 	if err != nil {
 		minioErr := s3.ToErrorResponse(err)
 		if minioErr.Code == noSuchKeyCode {
 			return false, nil
 		}
 		return false, err
 	}
-	if objInfo.Err != nil {
-		return false, nil
-	}
 	return true, nil
 }
 
 func UploadObject(putter ObjectPutter, bucketName, objKey string, reader io.Reader) error {
 	_, err := putter.PutObject(bucketName, objKey, reader, octetStream)
 	return err
 }
+
+// WaitForObject checks statter for the object at ${bucketName}/${objKey} right away, then at every tick, then once when the timeout is up.
+// Returns nil if it finds the object before or at timeout. Otherwise returns an error
+func WaitForObject(statter ObjectStatter, bucketName, objKey string, tick, timeout time.Duration) error {
+	noExist := errors.New("object doesn't exist")
+	checker := func() error {
+		exists, err := ObjectExists(statter, bucketName, objKey)
+		if err != nil {
+			return err
+		} else if exists {
+			return nil
+		} else {
+			return noExist
+		}
+	}
+	if err := checker(); err == nil {
+		return nil
+	}
+	ticker := time.NewTicker(tick)
+	defer ticker.Stop()
+	timer := time.NewTimer(timeout)
+	defer timer.Stop()
+	for {
+		select {
+		case <-ticker.C:
+			if err := checker(); err == nil {
+				return nil
+			}
+		case <-timer.C:
+			if err := checker(); err == nil {
+				return nil
+			}
+			return fmt.Errorf("Object %s/%s didn't exist after %s", bucketName, objKey, timeout)
+		}
+	}
+}

pkg/gitreceive/storage/object.go

@@ -5,6 +5,7 @@ import (
 	"io"
 	"strings"
 	"testing"
+	"time"
 
 	"github.com/arschles/assert"
 	s3 "github.com/minio/minio-go"
@@ -78,3 +79,26 @@ func TestUploadObjectFailure(t *testing.T) {
 	assert.Err(t, UploadObject(putter, bucketName, objKey, rdr), err)
 	assert.Equal(t, len(putter.Calls), 1, "number of calls to PutObject")
 }
+
+func TestWaitForObjectMissing(t *testing.T) {
+	statter := &FakeObjectStatter{
+		Fn: func(string, string) (s3.ObjectInfo, error) {
+			return s3.ObjectInfo{}, s3.ErrorResponse{Code: noSuchKeyCode}
+		},
+	}
+	err := WaitForObject(statter, bucketName, objKey, 1*time.Millisecond, 2*time.Millisecond)
+	assert.True(t, err != nil, "no error received when there should have been")
+	// it should make 1 call immediately, then calls at 1ms and 2ms
+	assert.Equal(t, len(statter.Calls), 3, "number of calls to the statter")
+}
+
+func TestWaitForObjectExists(t *testing.T) {
+	statter := &FakeObjectStatter{
+		Fn: func(string, string) (s3.ObjectInfo, error) {
+			return s3.ObjectInfo{}, nil
+		},
+	}
+	assert.NoErr(t, WaitForObject(statter, bucketName, objKey, 1*time.Millisecond, 2*time.Millisecond))
+	// it should make 1 call immediately, then immediateley succeed
+	assert.Equal(t, len(statter.Calls), 1, "number of calls to the statter")
+}