feat(storage): refactor to use transport-agnostic interface

internal/kokoro/continuous.sh

@@ -56,7 +56,8 @@ set -eo pipefail
 set -x
 
 # cd to project dir on Kokoro instance
-cd github/google-cloud-go
+# TODO(#6444): change this back to github/google-cloud-go before merging to main
+cd github/google-cloud-go-storage-refactor
 
 go version
 

storage/acl.go

@@ -21,7 +21,6 @@ import (
 
 	"cloud.google.com/go/internal/trace"
 	storagepb "cloud.google.com/go/storage/internal/apiv2/stubs"
-	"google.golang.org/api/googleapi"
 	raw "google.golang.org/api/storage/v1"
 )
 
@@ -121,111 +120,46 @@ func (a *ACLHandle) List(ctx context.Context) (rules []ACLRule, err error) {
 }
 
 func (a *ACLHandle) bucketDefaultList(ctx context.Context) ([]ACLRule, error) {
-	var acls *raw.ObjectAccessControls
-	var err error
-	req := a.c.raw.DefaultObjectAccessControls.List(a.bucket)
-	a.configureCall(ctx, req)
-	err = run(ctx, func() error {
-		acls, err = req.Do()
-		return err
-	}, a.retry, true, setRetryHeaderHTTP(req))
-	if err != nil {
-		return nil, err
-	}
-	return toObjectACLRules(acls.Items), nil
+	opts := makeStorageOpts(true, a.retry, a.userProject)
+	return a.c.tc.ListDefaultObjectACLs(ctx, a.bucket, opts...)
 }
 
 func (a *ACLHandle) bucketDefaultDelete(ctx context.Context, entity ACLEntity) error {
-	req := a.c.raw.DefaultObjectAccessControls.Delete(a.bucket, string(entity))
-	a.configureCall(ctx, req)
-
-	return run(ctx, func() error {
-		return req.Do()
-	}, a.retry, false, setRetryHeaderHTTP(req))
+	opts := makeStorageOpts(false, a.retry, a.userProject)
+	return a.c.tc.DeleteDefaultObjectACL(ctx, a.bucket, entity, opts...)
 }
 
 func (a *ACLHandle) bucketList(ctx context.Context) ([]ACLRule, error) {
-	var acls *raw.BucketAccessControls
-	var err error
-	req := a.c.raw.BucketAccessControls.List(a.bucket)
-	a.configureCall(ctx, req)
-	err = run(ctx, func() error {
-		acls, err = req.Do()
-		return err
-	}, a.retry, true, setRetryHeaderHTTP(req))
-	if err != nil {
-		return nil, err
-	}
-	return toBucketACLRules(acls.Items), nil
+	opts := makeStorageOpts(true, a.retry, a.userProject)
+	return a.c.tc.ListBucketACLs(ctx, a.bucket, opts...)
 }
 
 func (a *ACLHandle) bucketSet(ctx context.Context, entity ACLEntity, role ACLRole) error {
-	acl := &raw.BucketAccessControl{
-		Bucket: a.bucket,
-		Entity: string(entity),
-		Role:   string(role),
-	}
-	req := a.c.raw.BucketAccessControls.Update(a.bucket, string(entity), acl)
-	a.configureCall(ctx, req)
-	return run(ctx, func() error {
-		_, err := req.Do()
-		return err
-	}, a.retry, false, setRetryHeaderHTTP(req))
+	opts := makeStorageOpts(false, a.retry, a.userProject)
+	return a.c.tc.UpdateBucketACL(ctx, a.bucket, entity, role, opts...)
 }
 
 func (a *ACLHandle) bucketDelete(ctx context.Context, entity ACLEntity) error {
-	req := a.c.raw.BucketAccessControls.Delete(a.bucket, string(entity))
-	a.configureCall(ctx, req)
-	return run(ctx, func() error {
-		return req.Do()
-	}, a.retry, false, setRetryHeaderHTTP(req))
+	opts := makeStorageOpts(false, a.retry, a.userProject)
+	return a.c.tc.DeleteBucketACL(ctx, a.bucket, entity, opts...)
 }
 
 func (a *ACLHandle) objectList(ctx context.Context) ([]ACLRule, error) {
-	var acls *raw.ObjectAccessControls
-	var err error
-	req := a.c.raw.ObjectAccessControls.List(a.bucket, a.object)
-	a.configureCall(ctx, req)
-	err = run(ctx, func() error {
-		acls, err = req.Do()
-		return err
-	}, a.retry, true, setRetryHeaderHTTP(req))
-	if err != nil {
-		return nil, err
-	}
-	return toObjectACLRules(acls.Items), nil
+	opts := makeStorageOpts(true, a.retry, a.userProject)
+	return a.c.tc.ListObjectACLs(ctx, a.bucket, a.object, opts...)
 }
 
 func (a *ACLHandle) objectSet(ctx context.Context, entity ACLEntity, role ACLRole, isBucketDefault bool) error {
-	type setRequest interface {
-		Do(opts ...googleapi.CallOption) (*raw.ObjectAccessControl, error)
-		Header() http.Header
-	}
-
-	acl := &raw.ObjectAccessControl{
-		Bucket: a.bucket,
-		Entity: string(entity),
-		Role:   string(role),
-	}
-	var req setRequest
+	opts := makeStorageOpts(false, a.retry, a.userProject)
 	if isBucketDefault {
-		req = a.c.raw.DefaultObjectAccessControls.Update(a.bucket, string(entity), acl)
-	} else {
-		req = a.c.raw.ObjectAccessControls.Update(a.bucket, a.object, string(entity), acl)
+		return a.c.tc.UpdateDefaultObjectACL(ctx, a.bucket, entity, role, opts...)
 	}
-	a.configureCall(ctx, req)
-	return run(ctx, func() error {
-		_, err := req.Do()
-		return err
-	}, a.retry, false, setRetryHeaderHTTP(req))
+	return a.c.tc.UpdateObjectACL(ctx, a.bucket, a.object, entity, role, opts...)
 }
 
 func (a *ACLHandle) objectDelete(ctx context.Context, entity ACLEntity) error {
-	req := a.c.raw.ObjectAccessControls.Delete(a.bucket, a.object, string(entity))
-	a.configureCall(ctx, req)
-	return run(ctx, func() error {
-		return req.Do()
-	}, a.retry, false, setRetryHeaderHTTP(req))
+	opts := makeStorageOpts(false, a.retry, a.userProject)
+	return a.c.tc.DeleteObjectACL(ctx, a.bucket, a.object, entity, opts...)
 }
 
 func (a *ACLHandle) configureCall(ctx context.Context, call interface{ Header() http.Header }) {

storage/bucket.go

@@ -82,27 +82,11 @@ func (b *BucketHandle) Create(ctx context.Context, projectID string, attrs *Buck
 	ctx = trace.StartSpan(ctx, "cloud.google.com/go/storage.Bucket.Create")
 	defer func() { trace.EndSpan(ctx, err) }()
 
-	var bkt *raw.Bucket
-	if attrs != nil {
-		bkt = attrs.toRawBucket()
-	} else {
-		bkt = &raw.Bucket{}
-	}
-	bkt.Name = b.name
-	// If there is lifecycle information but no location, explicitly set
-	// the location. This is a GCS quirk/bug.
-	if bkt.Location == "" && bkt.Lifecycle != nil {
-		bkt.Location = "US"
-	}
-	req := b.c.raw.Buckets.Insert(projectID, bkt)
-	setClientHeader(req.Header())
-	if attrs != nil && attrs.PredefinedACL != "" {
-		req.PredefinedAcl(attrs.PredefinedACL)
-	}
-	if attrs != nil && attrs.PredefinedDefaultObjectACL != "" {
-		req.PredefinedDefaultObjectAcl(attrs.PredefinedDefaultObjectACL)
+	o := makeStorageOpts(true, b.retry, b.userProject)
+	if _, err := b.c.tc.CreateBucket(ctx, projectID, b.name, attrs, o...); err != nil {
+		return err
 	}
-	return run(ctx, func() error { _, err := req.Context(ctx).Do(); return err }, b.retry, true, setRetryHeaderHTTP(req))
+	return nil
 }
 
 // Delete deletes the Bucket.
@@ -2124,17 +2108,8 @@ func (it *ObjectIterator) fetch(pageSize int, pageToken string) (string, error)
 //
 // Note: The returned iterator is not safe for concurrent operations without explicit synchronization.
 func (c *Client) Buckets(ctx context.Context, projectID string) *BucketIterator {
-	it := &BucketIterator{
-		ctx:       ctx,
-		client:    c,
-		projectID: projectID,
-	}
-	it.pageInfo, it.nextFunc = iterator.NewPageInfo(
-		it.fetch,
-		func() int { return len(it.buckets) },
-		func() interface{} { b := it.buckets; it.buckets = nil; return b })
-
-	return it
+	o := makeStorageOpts(true, c.retry, "")
+	return c.tc.ListBuckets(ctx, projectID, o...)
 }
 
 // A BucketIterator is an iterator over BucketAttrs.
@@ -2145,7 +2120,6 @@ type BucketIterator struct {
 	Prefix string
 
 	ctx       context.Context
-	client    *Client
 	projectID string
 	buckets   []*BucketAttrs
 	pageInfo  *iterator.PageInfo
@@ -2171,36 +2145,6 @@ func (it *BucketIterator) Next() (*BucketAttrs, error) {
 // Note: This method is not safe for concurrent operations without explicit synchronization.
 func (it *BucketIterator) PageInfo() *iterator.PageInfo { return it.pageInfo }
 
-// TODO: When the transport-agnostic client interface is integrated into the Veneer,
-// this method should be removed, and the iterator should be initialized by the
-// transport-specific client implementations.
-func (it *BucketIterator) fetch(pageSize int, pageToken string) (token string, err error) {
-	req := it.client.raw.Buckets.List(it.projectID)
-	setClientHeader(req.Header())
-	req.Projection("full")
-	req.Prefix(it.Prefix)
-	req.PageToken(pageToken)
-	if pageSize > 0 {
-		req.MaxResults(int64(pageSize))
-	}
-	var resp *raw.Buckets
-	err = run(it.ctx, func() error {
-		resp, err = req.Context(it.ctx).Do()
-		return err
-	}, it.client.retry, true, setRetryHeaderHTTP(req))
-	if err != nil {
-		return "", err
-	}
-	for _, item := range resp.Items {
-		b, err := newBucket(item)
-		if err != nil {
-			return "", err
-		}
-		it.buckets = append(it.buckets, b)
-	}
-	return resp.NextPageToken, nil
-}
-
 // RPO (Recovery Point Objective) configures the turbo replication feature. See
 // https://cloud.google.com/storage/docs/managing-turbo-replication for more information.
 type RPO int

storage/client.go

@@ -44,7 +44,7 @@ type storageClient interface {
 	// Top-level methods.
 
 	GetServiceAccount(ctx context.Context, project string, opts ...storageOption) (string, error)
-	CreateBucket(ctx context.Context, project string, attrs *BucketAttrs, opts ...storageOption) (*BucketAttrs, error)
+	CreateBucket(ctx context.Context, project, bucket string, attrs *BucketAttrs, opts ...storageOption) (*BucketAttrs, error)
 	ListBuckets(ctx context.Context, project string, opts ...storageOption) *BucketIterator
 	Close() error
 
@@ -162,6 +162,20 @@ func callSettings(defaults *settings, opts ...storageOption) *settings {
 	return &cs
 }
 
+// makeStorageOpts is a helper for generating a set of storageOption based on
+// idempotency, retryConfig, and userProject. All top-level client operations
+// will generally have to pass these options through the interface.
+func makeStorageOpts(isIdempotent bool, retry *retryConfig, userProject string) []storageOption {
+	opts := []storageOption{idempotent(isIdempotent)}
+	if retry != nil {
+		opts = append(opts, withRetryConfig(retry))
+	}
+	if userProject != "" {
+		opts = append(opts, withUserProject(userProject))
+	}
+	return opts
+}
+
 // storageOption is the transport-agnostic call option for the storageClient
 // interface.
 type storageOption interface {
@@ -281,7 +295,6 @@ type composeObjectRequest struct {
 	dstObject     destinationObject
 	srcs          []sourceObject
 	predefinedACL string
-	encryptionKey []byte
 	sendCRC32C    bool
 }
 
@@ -313,5 +326,6 @@ type rewriteObjectResponse struct {
 	resource *ObjectAttrs
 	done     bool
 	written  int64
+	size     int64
 	token    string
 }