feat(storage): add rpo (turbo replication) support

storage/bucket.go

@@ -461,6 +461,12 @@ type BucketAttrs struct {
 	// The project number of the project the bucket belongs to.
 	// This field is read-only.
 	ProjectNumber uint64
+
+	// RPO configures the Recovery Point Objective (RPO) policy of the bucket.
+	// Set to RPOAsyncTurbo to turn on Turbo Replication for a bucket.
+	// See https://cloud.google.com/storage/docs/managing-turbo-replication for
+	// more information.
+	RPO RPO
 }
 
 // BucketPolicyOnly is an alias for UniformBucketLevelAccess.
@@ -728,6 +734,7 @@ func newBucket(b *raw.Bucket) (*BucketAttrs, error) {
 		Etag:                     b.Etag,
 		LocationType:             b.LocationType,
 		ProjectNumber:            b.ProjectNumber,
+		RPO:                      toRPO(b),
 	}, nil
 }
 
@@ -780,6 +787,7 @@ func (b *BucketAttrs) toRawBucket() *raw.Bucket {
 		Logging:          b.Logging.toRawBucketLogging(),
 		Website:          b.Website.toRawBucketWebsite(),
 		IamConfiguration: bktIAM,
+		Rpo:              b.RPO.String(),
 	}
 }
 
@@ -889,6 +897,12 @@ type BucketAttrsToUpdate struct {
 	// See https://cloud.google.com/storage/docs/json_api/v1/buckets/patch.
 	PredefinedDefaultObjectACL string
 
+	// RPO configures the Recovery Point Objective (RPO) policy of the bucket.
+	// Set to RPOAsyncTurbo to turn on Turbo Replication for a bucket.
+	// See https://cloud.google.com/storage/docs/managing-turbo-replication for
+	// more information.
+	RPO RPO
+
 	setLabels    map[string]string
 	deleteLabels map[string]bool
 }
@@ -1001,7 +1015,10 @@ func (ua *BucketAttrsToUpdate) toRawBucket() *raw.Bucket {
 		rb.DefaultObjectAcl = nil
 		rb.ForceSendFields = append(rb.ForceSendFields, "DefaultObjectAcl")
 	}
+
 	rb.StorageClass = ua.StorageClass
+	rb.Rpo = ua.RPO.String()
+
 	if ua.setLabels != nil || ua.deleteLabels != nil {
 		rb.Labels = map[string]string{}
 		for k, v := range ua.setLabels {
@@ -1347,6 +1364,20 @@ func toPublicAccessPrevention(b *raw.BucketIamConfiguration) PublicAccessPrevent
 	}
 }
 
+func toRPO(b *raw.Bucket) RPO {
+	if b == nil {
+		return RPODefault
+	}
+	switch b.Rpo {
+	case rpoDefault:
+		return RPODefault
+	case rpoAsyncTurbo:
+		return RPOAsyncTurbo
+	default:
+		return RPODefault
+	}
+}
+
 // Objects returns an iterator over the objects in the bucket that match the
 // Query q. If q is nil, no filtering is done. Objects will be iterated over
 // lexicographically by name.
@@ -1534,3 +1565,29 @@ func (it *BucketIterator) fetch(pageSize int, pageToken string) (token string, e
 	}
 	return resp.NextPageToken, nil
 }
+
+type RPO int
+
+const (
+	// See: https://cloud.google.com/storage/docs/managing-turbo-replication
+
+	// RPODefault is used to reset RPO on an existing bucket with RPO set to RPOAsyncTurbo.
+	// Otherwise RPODefault is equivalent to the RPO field being absent, and is always ignored
+	RPODefault RPO = iota
+	// Set RPO to RPOAsyncTurbo to enable Turbo Replication on a bucket
+	RPOAsyncTurbo
+
+	rpoDefault    string = "DEFAULT"
+	rpoAsyncTurbo        = "ASYNC_TURBO"
+)
+
+func (rpo RPO) String() string {
+	switch rpo {
+	case RPODefault:
+		return rpoDefault
+	case RPOAsyncTurbo:
+		return rpoAsyncTurbo
+	default:
+		return rpoDefault
+	}
+}

storage/bucket_test.go

@@ -44,6 +44,7 @@ func TestBucketAttrsToRawBucket(t *testing.T) {
 		UniformBucketLevelAccess: UniformBucketLevelAccess{Enabled: true},
 		PublicAccessPrevention:   PublicAccessPreventionEnforced,
 		VersioningEnabled:        false,
+		RPO:                      RPOAsyncTurbo,
 		// should be ignored:
 		MetaGeneration: 39,
 		Created:        time.Now(),
@@ -125,6 +126,7 @@ func TestBucketAttrsToRawBucket(t *testing.T) {
 			PublicAccessPrevention: "enforced",
 		},
 		Versioning: nil, // ignore VersioningEnabled if false
+		Rpo:        rpoAsyncTurbo,
 		Labels:     map[string]string{"label": "value"},
 		Cors: []*raw.BucketCors{
 			{
@@ -278,6 +280,14 @@ func TestBucketAttrsToRawBucket(t *testing.T) {
 		t.Errorf(msg)
 	}
 
+	// Test that setting RPO to default is propagated in the proto.
+	attrs.RPO = RPODefault
+	got = attrs.toRawBucket()
+	want.Rpo = rpoDefault
+	if msg := testutil.Diff(got, want); msg != "" {
+		t.Errorf(msg)
+	}
+
 	// Re-enable UBLA and confirm that it does not affect the PAP setting.
 	attrs.UniformBucketLevelAccess = UniformBucketLevelAccess{Enabled: true}
 	got = attrs.toRawBucket()