chore(storage): implement Read with gRPC

storage/go.mod

@@ -12,4 +12,5 @@ require (
 	google.golang.org/api v0.52.0
 	google.golang.org/genproto v0.0.0-20210805201207-89edb61ffb67
 	google.golang.org/grpc v1.39.1
+	google.golang.org/protobuf v1.27.1
 )

storage/integration_test.go

@@ -28,6 +28,7 @@ import (
 	"io"
 	"io/ioutil"
 	"log"
+	"math"
 	"math/rand"
 	"mime/multipart"
 	"net/http"
@@ -61,6 +62,7 @@ const (
 	// TODO(jba): move to testutil, factor out from firestore/integration_test.go.
 	envFirestoreProjID     = "GCLOUD_TESTS_GOLANG_FIRESTORE_PROJECT_ID"
 	envFirestorePrivateKey = "GCLOUD_TESTS_GOLANG_FIRESTORE_KEY"
+	grpcBucket             = "golang-grpc-test-integration"
 )
 
 var (
@@ -203,6 +205,22 @@ func testConfig(ctx context.Context, t *testing.T) *Client {
 	return client
 }
 
+// testConfigGPRC returns a gRPC-based client to access GCS. testConfigGRPC
+// skips the curent test when being run in Short mode.
+func testConfigGRPC(ctx context.Context, t *testing.T) (gc *Client) {
+	if testing.Short() {
+		t.Skip("Integration tests skipped in short mode")
+	}
+
+	var err error
+	gc, err = newHybridClient(ctx, nil)
+	if err != nil {
+		log.Fatalf("newClientWithGRPC: %v", err)
+	}
+
+	return
+}
+
 // config is like testConfig, but it doesn't need a *testing.T.
 func config(ctx context.Context) *Client {
 	ts := testutil.TokenSource(ctx, ScopeFullControl)
@@ -744,6 +762,197 @@ func TestIntegration_ObjectsRangeReader(t *testing.T) {
 	}
 }
 
+func TestIntegration_ObjectReadGRPC(t *testing.T) {
+	ctx := context.Background()
+
+	// Create an HTTP client to upload test data and a gRPC client to test with.
+	hc := testConfig(ctx, t)
+	defer hc.Close()
+	gc := testConfigGRPC(ctx, t)
+	defer gc.Close()
+
+	content := []byte("Hello, world this is a grpc request")
+
+	// Upload test data.
+	name := uidSpace.New()
+	ho := hc.Bucket(grpcBucket).Object(name)
+	if err := writeObject(ctx, ho, "text/plain", content); err != nil {
+		t.Fatal(err)
+	}
+	defer func() {
+		if err := ho.Delete(ctx); err != nil {
+			log.Printf("failed to delete test object: %v", err)
+		}
+	}()
+
+	obj := gc.Bucket(grpcBucket).Object(name)
+
+	// Using a negative length to indicate reading to the end.
+	r, err := obj.NewRangeReader(ctx, 0, -1)
+	if err != nil {
+		t.Fatal(err)
+	}
+	defer r.Close()
+
+	b := new(bytes.Buffer)
+	b.Grow(len(content))
+
+	n, err := io.Copy(b, r)
+	if err != nil {
+		t.Fatal(err)
+	}
+	if n == 0 {
+		t.Fatal("Expected to have read more than 0 bytes")
+	}
+
+	got := b.String()
+	want := string(content)
+	if diff := cmp.Diff(got, want); diff != "" {
+		t.Errorf("got(-),want(+):\n%s", diff)
+	}
+}
+
+func TestIntegration_ObjectReadChunksGRPC(t *testing.T) {
+	ctx := context.Background()
+
+	// Create an HTTP client to upload test data and a gRPC client to test with.
+	hc := testConfig(ctx, t)
+	defer hc.Close()
+	gc := testConfigGRPC(ctx, t)
+	defer gc.Close()
+
+	// Use a larger blob to test chunking logic. This is a little over 5MB.
+	content := bytes.Repeat([]byte("a"), 5<<20)
+
+	// Upload test data.
+	name := uidSpace.New()
+	ho := hc.Bucket(grpcBucket).Object(name)
+	if err := writeObject(ctx, ho, "text/plain", content); err != nil {
+		t.Fatal(err)
+	}
+	defer func() {
+		if err := ho.Delete(ctx); err != nil {
+			log.Printf("failed to delete test object: %v", err)
+		}
+	}()
+
+	obj := gc.Bucket(grpcBucket).Object(name)
+
+	r, err := obj.NewReader(ctx)
+	if err != nil {
+		t.Fatal(err)
+	}
+	defer r.Close()
+
+	bufSize := len(content)
+	buf := make([]byte, bufSize)
+
+	// Read in smaller chunks, offset to provoke reading across a Recv boundary.
+	chunk := 4<<10 + 1234
+	offset := 0
+	for {
+		end := math.Min(float64(offset+chunk), float64(bufSize))
+		n, err := r.Read(buf[offset:int(end)])
+		if err == io.EOF {
+			break
+		}
+		if err != nil {
+			t.Fatal(err)
+		}
+		offset += n
+	}
+
+	// TODO: Verify content with the checksums.
+}
+
+func TestIntegration_ObjectReadRelativeToEndGRPC(t *testing.T) {
+	ctx := context.Background()
+
+	// Create an HTTP client to upload test data and a gRPC client to test with.
+	hc := testConfig(ctx, t)
+	defer hc.Close()
+	gc := testConfigGRPC(ctx, t)
+	defer gc.Close()
+
+	content := []byte("Hello, world this is a grpc request")
+
+	// Upload test data.
+	name := uidSpace.New()
+	ho := hc.Bucket(grpcBucket).Object(name)
+	if err := writeObject(ctx, ho, "text/plain", content); err != nil {
+		t.Fatal(err)
+	}
+	defer func() {
+		if err := ho.Delete(ctx); err != nil {
+			log.Printf("failed to delete test object: %v", err)
+		}
+	}()
+
+	obj := gc.Bucket(grpcBucket).Object(name)
+
+	offset := 7
+	// Using a negative offset to start reading relative to the end of the
+	// object, and length to indicate reading to the end.
+	r, err := obj.NewRangeReader(ctx, int64(offset*-1), -1)
+	if err != nil {
+		t.Fatal(err)
+	}
+	defer r.Close()
+
+	b := new(bytes.Buffer)
+	b.Grow(offset)
+
+	n, err := io.Copy(b, r)
+	if err != nil {
+		t.Fatal(err)
+	}
+	if n == 0 {
+		t.Fatal("Expected to have read more than 0 bytes")
+	}
+
+	got := b.String()
+	want := string(content[len(content)-offset:])
+	if diff := cmp.Diff(got, want); diff != "" {
+		t.Errorf("got(-),want(+):\n%s", diff)
+	}
+}
+
+func TestIntegration_ConditionalDownloadGRPC(t *testing.T) {
+	ctx := context.Background()
+
+	// Create an HTTP client to upload test data and a gRPC client to test with.
+	hc := testConfig(ctx, t)
+	defer hc.Close()
+	gc := testConfigGRPC(ctx, t)
+	defer gc.Close()
+	h := testHelper{t}
+
+	o := hc.Bucket(grpcBucket).Object("condread")
+	defer o.Delete(ctx)
+
+	wc := o.NewWriter(ctx)
+	wc.ContentType = "text/plain"
+	h.mustWrite(wc, []byte("foo"))
+
+	gen := wc.Attrs().Generation
+	metaGen := wc.Attrs().Metageneration
+
+	obj := gc.Bucket(grpcBucket).Object(o.ObjectName())
+
+	if _, err := obj.Generation(gen + 1).NewReader(ctx); err == nil {
+		t.Fatalf("Unexpected successful download with nonexistent Generation")
+	}
+	if _, err := obj.If(Conditions{MetagenerationMatch: metaGen + 1}).NewReader(ctx); err == nil {
+		t.Fatalf("Unexpected successful download with failed preconditions IfMetaGenerationMatch")
+	}
+	if _, err := obj.If(Conditions{GenerationMatch: gen + 1}).NewReader(ctx); err == nil {
+		t.Fatalf("Unexpected successful download with failed preconditions IfGenerationMatch")
+	}
+	if _, err := obj.If(Conditions{GenerationMatch: gen}).NewReader(ctx); err != nil {
+		t.Fatalf("Download failed: %v", err)
+	}
+}
+
 func TestIntegration_ConditionalDownload(t *testing.T) {
 	ctx := context.Background()
 	client := testConfig(ctx, t)

storage/invoke.go

@@ -23,6 +23,8 @@ import (
 	"cloud.google.com/go/internal"
 	gax "github.com/googleapis/gax-go/v2"
 	"google.golang.org/api/googleapi"
+	"google.golang.org/grpc/codes"
+	"google.golang.org/grpc/status"
 )
 
 // runWithRetry calls the function until it returns nil or a non-retryable error, or
@@ -64,6 +66,13 @@ func shouldRetry(err error) bool {
 			return true
 		}
 	}
+	// HTTP 429, 502, 503, and 504 all map to gRPC UNAVAILABLE per
+	// https://grpc.github.io/grpc/core/md_doc_http-grpc-status-mapping.html.
+	//
+	// This is only necessary for the experimental gRPC-based media operations.
+	if st, ok := status.FromError(err); ok && st.Code() == codes.Unavailable {
+		return true
+	}
 	// Unwrap is only supported in go1.13.x+
 	if e, ok := err.(interface{ Unwrap() error }); ok {
 		return shouldRetry(e.Unwrap())