Merge pull request #1209 from percona/PBM-1093-azure-improvements

PBM-1093: Agents stop processing requests in case PBM runs with Azure

Author: boris-ilijic

packaging/conf/pbm-conf-reference.yml

@@ -69,8 +69,8 @@
 ## Retry upload configuration options.
 #      retryer:
 #        numMaxRetries: 3
-#        minRetryDelay: 30
-#        maxRetryDelay: 5
+#        minRetryDelay: 30ms
+#        maxRetryDelay: 300s
 #
 ## The maximum object size that will be stored on the storage
 #      maxObjSizeGB: 5018
@@ -183,6 +183,12 @@
 #      credentials:
 #        key:
 #
+## Retry upload configuration options.
+#      retryer:
+#        numMaxRetries: 3
+#        minRetryDelay: 800ms
+#        maxRetryDelay: 60s
+#
 ## The maximum object size that will be stored on the storage
 #      maxObjSizeGB: 194560
 

pbm/config/config.go

@@ -343,10 +343,10 @@ func (s *StorageConf) Cast() error {
 		return s.Minio.Cast()
 	case storage.OSS:
 		return s.OSS.Cast()
+	case storage.Azure:
+		return s.Azure.Cast()
 	case storage.GCS:
 		return nil
-	case storage.Azure: // noop
-		return nil
 	case storage.Blackhole: // noop
 		return nil
 	}

pbm/storage/azure/azure.go

@@ -2,12 +2,9 @@ package azure
 
 import (
 	"context"
-	"fmt"
 	"io"
-	"maps"
 	"net/http"
 	"path"
-	"reflect"
 	"runtime"
 	"strings"
 	"time"
@@ -25,139 +22,60 @@ import (
 const (
 	BlobURL = "https://%s.blob.core.windows.net"
 
-	defaultUploadBuff    = 10 << 20 // 10Mb
-	defaultUploadMaxBuff = 5
+	defaultUploadBuff = 10 << 20 // 10Mb
 
-	defaultRetries = 10
+	defaultMaxRetries    = 3
+	defaultMinRetryDelay = 800 * time.Millisecond
+	defaultMaxRetryDelay = 60 * time.Second
 
 	maxBlocks = 50_000
 
 	defaultMaxObjSizeGB = 194560 // 190 TB
 )
 
-//nolint:lll
-type Config struct {
-	Account        string            `bson:"account" json:"account,omitempty" yaml:"account,omitempty"`
-	Container      string            `bson:"container" json:"container,omitempty" yaml:"container,omitempty"`
-	EndpointURL    string            `bson:"endpointUrl" json:"endpointUrl,omitempty" yaml:"endpointUrl,omitempty"`
-	EndpointURLMap map[string]string `bson:"endpointUrlMap,omitempty" json:"endpointUrlMap,omitempty" yaml:"endpointUrlMap,omitempty"`
-	Prefix         string            `bson:"prefix" json:"prefix,omitempty" yaml:"prefix,omitempty"`
-	Credentials    Credentials       `bson:"credentials" json:"-" yaml:"credentials"`
-	MaxObjSizeGB   *float64          `bson:"maxObjSizeGB,omitempty" json:"maxObjSizeGB,omitempty" yaml:"maxObjSizeGB,omitempty"`
-}
-
-func (cfg *Config) Clone() *Config {
-	if cfg == nil {
-		return nil
-	}
-
-	rv := *cfg
-	rv.EndpointURLMap = maps.Clone(cfg.EndpointURLMap)
-	if cfg.MaxObjSizeGB != nil {
-		v := *cfg.MaxObjSizeGB
-		rv.MaxObjSizeGB = &v
-	}
-	return &rv
-}
-
-func (cfg *Config) Equal(other *Config) bool {
-	if cfg == nil || other == nil {
-		return cfg == other
-	}
-
-	if cfg.Account != other.Account {
-		return false
-	}
-	if cfg.Container != other.Container {
-		return false
-	}
-	if cfg.EndpointURL != other.EndpointURL {
-		return false
-	}
-	if !maps.Equal(cfg.EndpointURLMap, other.EndpointURLMap) {
-		return false
-	}
-	if cfg.Prefix != other.Prefix {
-		return false
-	}
-	if cfg.Credentials.Key != other.Credentials.Key {
-		return false
-	}
-	if !reflect.DeepEqual(cfg.MaxObjSizeGB, other.MaxObjSizeGB) {
-		return false
-	}
-
-	return true
-}
-
-// IsSameStorage identifies the same instance of the Azure storage.
-func (cfg *Config) IsSameStorage(other *Config) bool {
-	if cfg == nil || other == nil {
-		return cfg == other
-	}
-
-	if cfg.Account != other.Account {
-		return false
-	}
-	if cfg.Container != other.Container {
-		return false
-	}
-	if cfg.Prefix != other.Prefix {
-		return false
-	}
-	return true
-}
-
-// resolveEndpointURL returns endpoint url based on provided
-// EndpointURL or associated EndpointURLMap configuration fields.
-// If specified EndpointURLMap overrides EndpointURL field.
-func (cfg *Config) resolveEndpointURL(node string) string {
-	ep := cfg.EndpointURL
-	if epm, ok := cfg.EndpointURLMap[node]; ok {
-		ep = epm
-	}
-	if ep == "" {
-		ep = fmt.Sprintf(BlobURL, cfg.Account)
-	}
-	return ep
-}
-
-func (cfg *Config) GetMaxObjSizeGB() float64 {
-	if cfg.MaxObjSizeGB != nil && *cfg.MaxObjSizeGB > 0 {
-		return *cfg.MaxObjSizeGB
-	}
-	return defaultMaxObjSizeGB
-}
-
-type Credentials struct {
-	Key string `bson:"key" json:"key,omitempty" yaml:"key,omitempty"`
-}
-
 type Blob struct {
-	opts *Config
+	cfg  *Config
 	node string
 	log  log.LogEvent
-	// url  *url.URL
-	c *azblob.Client
+	c    *azblob.Client
 }
 
-func New(opts *Config, node string, l log.LogEvent) (storage.Storage, error) {
+func New(cfg *Config, node string, l log.LogEvent) (storage.Storage, error) {
+	err := cfg.Cast()
+	if err != nil {
+		return nil, errors.Wrap(err, "set defaults")
+	}
 	if l == nil {
 		l = log.DiscardEvent
 	}
 	b := &Blob{
-		opts: opts,
+		cfg:  cfg,
 		node: node,
 		log:  l,
 	}
 
-	var err error
 	b.c, err = b.client()
 	if err != nil {
 		return nil, errors.Wrap(err, "init container")
 	}
 
-	return storage.NewSplitMergeMW(b, opts.GetMaxObjSizeGB()), b.ensureContainer()
+	return storage.NewSplitMergeMW(b, cfg.GetMaxObjSizeGB()), nil
+}
+
+func (b *Blob) client() (*azblob.Client, error) {
+	cred, err := azblob.NewSharedKeyCredential(b.cfg.Account, b.cfg.Credentials.Key)
+	if err != nil {
+		return nil, errors.Wrap(err, "create credentials")
+	}
+
+	opts := &azblob.ClientOptions{}
+	opts.Retry = policy.RetryOptions{
+		MaxRetries:    b.cfg.Retryer.NumMaxRetries,
+		RetryDelay:    b.cfg.Retryer.MinRetryDelay,
+		MaxRetryDelay: b.cfg.Retryer.MaxRetryDelay,
+	}
+	epURL := b.cfg.resolveEndpointURL(b.node)
+	return azblob.NewClientWithSharedKeyCredential(epURL, cred, opts)
 }
 
 func (*Blob) Type() storage.Type {
@@ -180,18 +98,15 @@ func (b *Blob) Save(name string, data io.Reader, options ...storage.Option) erro
 		}
 	}
 
-	cc := runtime.NumCPU() / 2
-	if cc == 0 {
-		cc = 1
-	}
+	cc := max(runtime.NumCPU()/2, 1)
 
 	if b.log != nil && opts.UseLogger {
 		b.log.Debug("BufferSize is set to %d (~%dMb) | %d", bufsz, bufsz>>20, opts.Size)
 	}
 
 	_, err := b.c.UploadStream(context.TODO(),
-		b.opts.Container,
-		path.Join(b.opts.Prefix, name),
+		b.cfg.Container,
+		path.Join(b.cfg.Prefix, name),
 		data,
 		&azblob.UploadStreamOptions{
 			BlockSize:   int64(bufsz),
@@ -202,13 +117,13 @@ func (b *Blob) Save(name string, data io.Reader, options ...storage.Option) erro
 }
 
 func (b *Blob) List(prefix, suffix string) ([]storage.FileInfo, error) {
-	prfx := path.Join(b.opts.Prefix, prefix)
+	prfx := path.Join(b.cfg.Prefix, prefix)
 
 	if prfx != "" && !strings.HasSuffix(prfx, "/") {
 		prfx += "/"
 	}
 
-	pager := b.c.NewListBlobsFlatPager(b.opts.Container, &azblob.ListBlobsFlatOptions{
+	pager := b.c.NewListBlobsFlatPager(b.cfg.Container, &azblob.ListBlobsFlatOptions{
 		Prefix: &prfx,
 	})
 
@@ -252,8 +167,8 @@ func (b *Blob) FileStat(name string) (storage.FileInfo, error) {
 	inf := storage.FileInfo{}
 
 	p, err := b.c.ServiceClient().
-		NewContainerClient(b.opts.Container).
-		NewBlockBlobClient(path.Join(b.opts.Prefix, name)).
+		NewContainerClient(b.cfg.Container).
+		NewBlockBlobClient(path.Join(b.cfg.Prefix, name)).
 		GetProperties(context.TODO(), nil)
 	if err != nil {
 		if isNotFound(err) {
@@ -275,8 +190,8 @@ func (b *Blob) FileStat(name string) (storage.FileInfo, error) {
 }
 
 func (b *Blob) Copy(src, dst string) error {
-	to := b.c.ServiceClient().NewContainerClient(b.opts.Container).NewBlockBlobClient(path.Join(b.opts.Prefix, dst))
-	from := b.c.ServiceClient().NewContainerClient(b.opts.Container).NewBlockBlobClient(path.Join(b.opts.Prefix, src))
+	to := b.c.ServiceClient().NewContainerClient(b.cfg.Container).NewBlockBlobClient(path.Join(b.cfg.Prefix, dst))
+	from := b.c.ServiceClient().NewContainerClient(b.cfg.Container).NewBlockBlobClient(path.Join(b.cfg.Prefix, src))
 	r, err := to.StartCopyFromURL(context.TODO(), from.BlobClient().URL(), nil)
 	if err != nil {
 		return errors.Wrap(err, "start copy")
@@ -316,7 +231,7 @@ func (b *Blob) DownloadStat() storage.DownloadStat {
 }
 
 func (b *Blob) SourceReader(name string) (io.ReadCloser, error) {
-	o, err := b.c.DownloadStream(context.TODO(), b.opts.Container, path.Join(b.opts.Prefix, name), nil)
+	o, err := b.c.DownloadStream(context.TODO(), b.cfg.Container, path.Join(b.cfg.Prefix, name), nil)
 	if err != nil {
 		if isNotFound(err) {
 			return nil, storage.ErrNotExist
@@ -335,7 +250,7 @@ func (b *Blob) SourceReader(name string) (io.ReadCloser, error) {
 }
 
 func (b *Blob) Delete(name string) error {
-	_, err := b.c.DeleteBlob(context.TODO(), b.opts.Container, path.Join(b.opts.Prefix, name), nil)
+	_, err := b.c.DeleteBlob(context.TODO(), b.cfg.Container, path.Join(b.cfg.Prefix, name), nil)
 	if err != nil {
 		if isNotFound(err) {
 			return storage.ErrNotExist
@@ -346,36 +261,6 @@ func (b *Blob) Delete(name string) error {
 	return nil
 }
 
-func (b *Blob) ensureContainer() error {
-	_, err := b.c.ServiceClient().NewContainerClient(b.opts.Container).GetProperties(context.TODO(), nil)
-	// container already exists
-	if err == nil {
-		return nil
-	}
-
-	var stgErr *azcore.ResponseError
-	if errors.As(err, &stgErr) && stgErr.StatusCode != http.StatusNotFound {
-		return errors.Wrap(err, "check container")
-	}
-
-	_, err = b.c.CreateContainer(context.TODO(), b.opts.Container, nil)
-	return err
-}
-
-func (b *Blob) client() (*azblob.Client, error) {
-	cred, err := azblob.NewSharedKeyCredential(b.opts.Account, b.opts.Credentials.Key)
-	if err != nil {
-		return nil, errors.Wrap(err, "create credentials")
-	}
-
-	opts := &azblob.ClientOptions{}
-	opts.Retry = policy.RetryOptions{
-		MaxRetries: defaultRetries,
-	}
-	epURL := b.opts.resolveEndpointURL(b.node)
-	return azblob.NewClientWithSharedKeyCredential(epURL, cred, opts)
-}
-
 func isNotFound(err error) bool {
 	var stgErr *azcore.ResponseError
 	if errors.As(err, &stgErr) {