PBM-1484 GCP SDK (#1096)

* add gcs
* add type
* update error handling
* add google storage sdk
* use iterator
* update mod file
* add chunk size
* add retryer
* add credentials check
* add multiplier

---------

Co-authored-by: github-actions[bot] <41898282+github-actions[bot]@users.noreply.github.com>

Author: veceraj

go.mod

@@ -3,6 +3,7 @@ module github.com/percona/percona-backup-mongodb
 go 1.22
 
 require (
+	cloud.google.com/go/storage v1.38.0
 	github.com/Azure/azure-sdk-for-go/sdk/azcore v1.10.0
 	github.com/Azure/azure-sdk-for-go/sdk/storage/azblob v1.3.1
 	github.com/aws/aws-sdk-go-v2 v1.33.0
@@ -16,6 +17,7 @@ require (
 	github.com/fsnotify/fsnotify v1.7.0
 	github.com/golang/snappy v0.0.4
 	github.com/google/uuid v1.6.0
+	github.com/googleapis/gax-go/v2 v2.12.3
 	github.com/klauspost/compress v1.17.11
 	github.com/klauspost/pgzip v1.2.6
 	github.com/mongodb/mongo-tools v0.0.0-20240723193119-837c2bc263f4
@@ -28,10 +30,15 @@ require (
 	go.mongodb.org/mongo-driver v1.17.1
 	golang.org/x/mod v0.19.0
 	golang.org/x/sync v0.11.0
+	google.golang.org/api v0.171.0
 	gopkg.in/yaml.v2 v2.4.0
 )
 
 require (
+	cloud.google.com/go v0.112.1 // indirect
+	cloud.google.com/go/compute v1.25.1 // indirect
+	cloud.google.com/go/compute/metadata v0.2.3 // indirect
+	cloud.google.com/go/iam v1.1.6 // indirect
 	dario.cat/mergo v1.0.0 // indirect
 	github.com/Azure/azure-sdk-for-go/sdk/internal v1.5.2 // indirect
 	github.com/Azure/go-ansiterm v0.0.0-20210617225240-d185dfc1b5a1 // indirect
@@ -62,6 +69,10 @@ require (
 	github.com/go-logr/stdr v1.2.2 // indirect
 	github.com/go-ole/go-ole v1.2.6 // indirect
 	github.com/gogo/protobuf v1.3.2 // indirect
+	github.com/golang/groupcache v0.0.0-20210331224755-41bb18bfe9da // indirect
+	github.com/golang/protobuf v1.5.4 // indirect
+	github.com/google/s2a-go v0.1.7 // indirect
+	github.com/googleapis/enterprise-certificate-proxy v0.3.2 // indirect
 	github.com/hashicorp/hcl v1.0.0 // indirect
 	github.com/inconshreveable/mousetrap v1.1.0 // indirect
 	github.com/jessevdk/go-flags v1.5.0 // indirect
@@ -98,6 +109,8 @@ require (
 	github.com/xdg-go/stringprep v1.0.4 // indirect
 	github.com/youmark/pkcs8 v0.0.0-20240726163527-a2c0da244d78 // indirect
 	github.com/yusufpapurcu/wmi v1.2.3 // indirect
+	go.opencensus.io v0.24.0 // indirect
+	go.opentelemetry.io/contrib/instrumentation/google.golang.org/grpc/otelgrpc v0.49.0 // indirect
 	go.opentelemetry.io/contrib/instrumentation/net/http/otelhttp v0.49.0 // indirect
 	go.opentelemetry.io/otel v1.24.0 // indirect
 	go.opentelemetry.io/otel/exporters/otlp/otlptrace/otlptracehttp v1.24.0 // indirect
@@ -109,9 +122,17 @@ require (
 	golang.org/x/crypto v0.33.0 // indirect
 	golang.org/x/exp v0.0.0-20240529005216-23cca8864a10 // indirect
 	golang.org/x/net v0.35.0 // indirect
+	golang.org/x/oauth2 v0.18.0 // indirect
 	golang.org/x/sys v0.30.0 // indirect
 	golang.org/x/term v0.29.0 // indirect
 	golang.org/x/text v0.22.0 // indirect
+	golang.org/x/time v0.5.0 // indirect
+	google.golang.org/appengine v1.6.8 // indirect
+	google.golang.org/genproto v0.0.0-20240213162025-012b6fc9bca9 // indirect
+	google.golang.org/genproto/googleapis/api v0.0.0-20240318140521-94a12d6c2237 // indirect
+	google.golang.org/genproto/googleapis/rpc v0.0.0-20240318140521-94a12d6c2237 // indirect
+	google.golang.org/grpc v1.64.1 // indirect
+	google.golang.org/protobuf v1.33.0 // indirect
 	gopkg.in/ini.v1 v1.67.0 // indirect
 	gopkg.in/yaml.v3 v3.0.1 // indirect
 )

go.sum

@@ -25,6 +25,7 @@ import (
 	"github.com/percona/percona-backup-mongodb/pbm/storage"
 	"github.com/percona/percona-backup-mongodb/pbm/storage/azure"
 	"github.com/percona/percona-backup-mongodb/pbm/storage/fs"
+	"github.com/percona/percona-backup-mongodb/pbm/storage/gcs"
 	"github.com/percona/percona-backup-mongodb/pbm/storage/s3"
 	"github.com/percona/percona-backup-mongodb/pbm/topo"
 )
@@ -145,6 +146,11 @@ func (c *Config) String() string {
 			c.Storage.Azure.Credentials.Key = "***"
 		}
 	}
+	if c.Storage.GCS != nil {
+		if c.Storage.GCS.Credentials.PrivateKey != "" {
+			c.Storage.GCS.Credentials.PrivateKey = "***"
+		}
+	}
 
 	b, err := yaml.Marshal(c)
 	if err != nil {
@@ -209,6 +215,7 @@ func (cfg *PITRConf) Clone() *PITRConf {
 type StorageConf struct {
 	Type       storage.Type  `bson:"type" json:"type" yaml:"type"`
 	S3         *s3.Config    `bson:"s3,omitempty" json:"s3,omitempty" yaml:"s3,omitempty"`
+	GCS        *gcs.Config   `bson:"gcs,omitempty" json:"gcs,omitempty" yaml:"gcs,omitempty"`
 	Azure      *azure.Config `bson:"azure,omitempty" json:"azure,omitempty" yaml:"azure,omitempty"`
 	Filesystem *fs.Config    `bson:"filesystem,omitempty" json:"filesystem,omitempty" yaml:"filesystem,omitempty"`
 }
@@ -229,6 +236,8 @@ func (s *StorageConf) Clone() *StorageConf {
 		rv.S3 = s.S3.Clone()
 	case storage.Azure:
 		rv.Azure = s.Azure.Clone()
+	case storage.GCS:
+		rv.GCS = s.GCS.Clone()
 	case storage.Blackhole: // no config
 	}
 
@@ -260,6 +269,8 @@ func (s *StorageConf) Cast() error {
 		return s.Filesystem.Cast()
 	case storage.S3:
 		return s.S3.Cast()
+	case storage.GCS:
+		return nil
 	case storage.Azure: // noop
 		return nil
 	case storage.Blackhole: // noop

pbm/config/config.go

@@ -0,0 +1,267 @@
+package gcs
+
+import (
+	"context"
+	"encoding/json"
+	"fmt"
+	"io"
+	"path"
+	"strings"
+	"time"
+
+	gcs "cloud.google.com/go/storage"
+	"github.com/googleapis/gax-go/v2"
+	"google.golang.org/api/iterator"
+	"google.golang.org/api/option"
+
+	"github.com/percona/percona-backup-mongodb/pbm/errors"
+	"github.com/percona/percona-backup-mongodb/pbm/log"
+	"github.com/percona/percona-backup-mongodb/pbm/storage"
+)
+
+type Config struct {
+	Bucket      string      `bson:"bucket" json:"bucket" yaml:"bucket"`
+	Prefix      string      `bson:"prefix" json:"prefix" yaml:"prefix"`
+	Credentials Credentials `bson:"credentials" json:"credentials" yaml:"credentials"`
+
+	// The maximum number of bytes that the Writer will attempt to send in a single request.
+	// https://pkg.go.dev/cloud.google.com/go/storage#Writer
+	ChunkSize *int `bson:"chunkSize,omitempty" json:"chunkSize,omitempty" yaml:"chunkSize,omitempty"`
+
+	Retryer *Retryer `bson:"retryer,omitempty" json:"retryer,omitempty" yaml:"retryer,omitempty"`
+}
+
+type Credentials struct {
+	ProjectID  string `bson:"projectId" json:"projectId,omitempty" yaml:"projectId,omitempty"`
+	PrivateKey string `bson:"privateKey" json:"privateKey,omitempty" yaml:"privateKey,omitempty"`
+}
+
+type Retryer struct {
+	// BackoffInitial is the initial value of the retry period.
+	// https://pkg.go.dev/github.com/googleapis/gax-go/[email protected]#Backoff.Initial
+	BackoffInitial time.Duration `bson:"backoffInitial" json:"backoffInitial" yaml:"backoffInitial"`
+
+	// BackoffMax is the maximum value of the retry period.
+	// https://pkg.go.dev/github.com/googleapis/gax-go/[email protected]#Backoff.Max
+	BackoffMax time.Duration `bson:"backoffMax" json:"backoffMax" yaml:"backoffMax"`
+
+	// BackoffMultiplier is the factor by which the retry period increases.
+	// https://pkg.go.dev/github.com/googleapis/gax-go/[email protected]#Backoff.Multiplier
+	BackoffMultiplier float64 `bson:"backoffMultiplier" json:"backoffMultiplier" yaml:"backoffMultiplier"`
+}
+
+type ServiceAccountCredentials struct {
+	Type                string `json:"type"`
+	ProjectID           string `json:"project_id"`
+	PrivateKey          string `json:"private_key"`
+	ClientEmail         string `json:"client_email"`
+	AuthURI             string `json:"auth_uri"`
+	TokenURI            string `json:"token_uri"`
+	UniverseDomain      string `json:"universe_domain"`
+	AuthProviderCertURL string `json:"auth_provider_x509_cert_url"`
+	ClientCertURL       string `json:"client_x509_cert_url"`
+}
+
+type GCS struct {
+	opts         *Config
+	bucketHandle *gcs.BucketHandle
+	log          log.LogEvent
+}
+
+func (cfg *Config) Clone() *Config {
+	if cfg == nil {
+		return nil
+	}
+
+	rv := *cfg
+	return &rv
+}
+
+func New(opts *Config, node string, l log.LogEvent) (*GCS, error) {
+	g := &GCS{
+		opts: opts,
+		log:  l,
+	}
+
+	cli, err := g.gcsClient()
+	if err != nil {
+		return nil, errors.Wrap(err, "GCS client")
+	}
+
+	bucketHandle := cli.Bucket(opts.Bucket)
+
+	if opts.Retryer != nil {
+		bucketHandle = bucketHandle.Retryer(
+			gcs.WithBackoff(gax.Backoff{
+				Initial:    opts.Retryer.BackoffInitial,
+				Max:        opts.Retryer.BackoffMax,
+				Multiplier: opts.Retryer.BackoffMultiplier,
+			}),
+
+			gcs.WithPolicy(gcs.RetryAlways),
+		)
+	}
+
+	g.bucketHandle = bucketHandle
+
+	return g, nil
+}
+
+func (*GCS) Type() storage.Type {
+	return storage.GCS
+}
+
+func (g *GCS) Save(name string, data io.Reader, size int64) error {
+	ctx := context.Background()
+
+	w := g.bucketHandle.Object(path.Join(g.opts.Prefix, name)).NewWriter(ctx)
+
+	if g.opts.ChunkSize != nil {
+		w.ChunkSize = *g.opts.ChunkSize
+	}
+
+	if _, err := io.Copy(w, data); err != nil {
+		return errors.Wrap(err, "save data")
+	}
+
+	if err := w.Close(); err != nil {
+		return errors.Wrap(err, "writer close")
+	}
+
+	return nil
+}
+
+func (g *GCS) SourceReader(name string) (io.ReadCloser, error) {
+	ctx := context.Background()
+
+	reader, err := g.bucketHandle.Object(path.Join(g.opts.Prefix, name)).NewReader(ctx)
+	if err != nil {
+		return nil, errors.Wrap(err, "object not found")
+	}
+
+	return reader, nil
+}
+
+func (g *GCS) FileStat(name string) (storage.FileInfo, error) {
+	ctx := context.Background()
+
+	attrs, err := g.bucketHandle.Object(path.Join(g.opts.Prefix, name)).Attrs(ctx)
+	if err != nil {
+		if errors.Is(err, gcs.ErrObjectNotExist) {
+			return storage.FileInfo{}, storage.ErrNotExist
+		}
+
+		return storage.FileInfo{}, errors.Wrap(err, "get properties")
+	}
+
+	inf := storage.FileInfo{
+		Name: attrs.Name,
+		Size: attrs.Size,
+	}
+
+	if inf.Size == 0 {
+		return inf, storage.ErrEmpty
+	}
+
+	return inf, nil
+}
+
+func (g *GCS) List(prefix, suffix string) ([]storage.FileInfo, error) {
+	ctx := context.Background()
+
+	prfx := path.Join(g.opts.Prefix, prefix)
+
+	if prfx != "" && !strings.HasSuffix(prfx, "/") {
+		prfx += "/"
+	}
+
+	query := &gcs.Query{
+		Prefix: prfx,
+	}
+
+	var files []storage.FileInfo
+	it := g.bucketHandle.Objects(ctx, query)
+	for {
+		attrs, err := it.Next()
+
+		if errors.Is(err, iterator.Done) {
+			break
+		}
+
+		if err != nil {
+			return nil, errors.Wrap(err, "list objects")
+		}
+
+		name := attrs.Name
+		name = strings.TrimPrefix(name, prfx)
+		if len(name) == 0 {
+			continue
+		}
+		if name[0] == '/' {
+			name = name[1:]
+		}
+
+		if suffix != "" && !strings.HasSuffix(name, suffix) {
+			continue
+		}
+
+		files = append(files, storage.FileInfo{
+			Name: name,
+			Size: attrs.Size,
+		})
+	}
+
+	return files, nil
+}
+
+func (g *GCS) Delete(name string) error {
+	ctx := context.Background()
+
+	err := g.bucketHandle.Object(path.Join(g.opts.Prefix, name)).Delete(ctx)
+	if err != nil {
+		if errors.Is(err, gcs.ErrObjectNotExist) {
+			return storage.ErrNotExist
+		}
+		return errors.Wrap(err, "delete object")
+	}
+
+	return nil
+}
+
+func (g *GCS) Copy(src, dst string) error {
+	ctx := context.Background()
+
+	srcObj := g.bucketHandle.Object(path.Join(g.opts.Prefix, src))
+	dstObj := g.bucketHandle.Object(path.Join(g.opts.Prefix, dst))
+
+	_, err := dstObj.CopierFrom(srcObj).Run(ctx)
+	return err
+}
+
+func (g *GCS) gcsClient() (*gcs.Client, error) {
+	ctx := context.Background()
+
+	if g.opts.Credentials.ProjectID == "" || g.opts.Credentials.PrivateKey == "" {
+		return nil, errors.New("projectID and privateKey are required for GCS credentials")
+	}
+
+	creds, err := json.Marshal(ServiceAccountCredentials{
+		Type:                "service_account",
+		ProjectID:           g.opts.Credentials.ProjectID,
+		PrivateKey:          g.opts.Credentials.PrivateKey,
+		ClientEmail:         fmt.Sprintf("service@%s.iam.gserviceaccount.com", g.opts.Credentials.ProjectID),
+		AuthURI:             "https://accounts.google.com/o/oauth2/auth",
+		TokenURI:            "https://oauth2.googleapis.com/token",
+		UniverseDomain:      "googleapis.com",
+		AuthProviderCertURL: "https://www.googleapis.com/oauth2/v1/certs",
+		ClientCertURL: fmt.Sprintf(
+			"https://www.googleapis.com/robot/v1/metadata/x509/%s.iam.gserviceaccount.com",
+			g.opts.Credentials.ProjectID,
+		),
+	})
+	if err != nil {
+		return nil, errors.Wrap(err, "marshal GCS credentials")
+	}
+
+	return gcs.NewClient(ctx, option.WithCredentialsJSON(creds))
+}