diff --git a/go.mod b/go.mod index c51b2e40..7a1e59a8 100644 --- a/go.mod +++ b/go.mod @@ -4,7 +4,8 @@ go 1.17 require ( cloud.google.com/go/storage v1.21.0 - github.com/Azure/azure-storage-blob-go v0.14.0 + github.com/Azure/azure-sdk-for-go/sdk/azcore v0.21.1 + github.com/Azure/azure-sdk-for-go/sdk/storage/azblob v0.3.0 github.com/GehirnInc/crypt v0.0.0-20200316065508-bb7000b8a962 github.com/alexedwards/argon2id v0.0.0-20211130144151-3585854a6387 github.com/aws/aws-sdk-go v1.43.2 @@ -67,7 +68,7 @@ require ( cloud.google.com/go v0.100.2 // indirect cloud.google.com/go/compute v1.3.0 // indirect cloud.google.com/go/iam v0.2.0 // indirect - github.com/Azure/azure-pipeline-go v0.2.3 // indirect + github.com/Azure/azure-sdk-for-go/sdk/internal v0.9.1 // indirect github.com/beorn7/perks v1.0.1 // indirect github.com/boombuler/barcode v1.0.1 // indirect github.com/cenkalti/backoff v2.2.1+incompatible // indirect @@ -99,7 +100,6 @@ require ( github.com/lufia/plan9stats v0.0.0-20211012122336-39d0f177ccd0 // indirect github.com/magiconair/properties v1.8.5 // indirect github.com/mattn/go-colorable v0.1.12 // indirect - github.com/mattn/go-ieproxy v0.0.3 // indirect github.com/mattn/go-isatty v0.0.14 // indirect github.com/matttproud/golang_protobuf_extensions v1.0.1 // indirect github.com/miekg/dns v1.1.46 // indirect diff --git a/go.sum b/go.sum index 07e75746..38c2dce1 100644 --- a/go.sum +++ b/go.sum @@ -81,43 +81,43 @@ contrib.go.opencensus.io/integrations/ocsql v0.1.7/go.mod h1:8DsSdjz3F+APR+0z0Wk dmitri.shuralyov.com/gpu/mtl v0.0.0-20190408044501-666a987793e9/go.mod h1:H6x//7gZCb22OMCxBHrMx7a5I7Hp++hsVxbQ4BYO7hU= github.com/Azure/azure-amqp-common-go/v3 v3.1.0/go.mod h1:PBIGdzcO1teYoufTKMcGibdKaYZv4avS+O6LNIp8bq0= github.com/Azure/azure-amqp-common-go/v3 v3.1.1/go.mod h1:YsDaPfaO9Ub2XeSKdIy2DfwuiQlHQCauHJwSqtrkECI= -github.com/Azure/azure-pipeline-go v0.2.3 h1:7U9HBg1JFK3jHl5qmo4CTZKFTVgMwdFHMVtCdfBE21U= github.com/Azure/azure-pipeline-go v0.2.3/go.mod h1:x841ezTBIMG6O3lAcl8ATHnsOPVl2bqk7S3ta6S6u4k= github.com/Azure/azure-sdk-for-go v51.1.0+incompatible/go.mod h1:9XXNKU+eRnpl9moKnB4QOLf1HestfXbmab5FXxiDBjc= +github.com/Azure/azure-sdk-for-go v57.0.0+incompatible h1:isVki3PbIFrwKvKdVP1byxo73/pt+Nn174YxW1k4PNw= github.com/Azure/azure-sdk-for-go v57.0.0+incompatible/go.mod h1:9XXNKU+eRnpl9moKnB4QOLf1HestfXbmab5FXxiDBjc= +github.com/Azure/azure-sdk-for-go/sdk/azcore v0.21.1 h1:qoVeMsc9/fh/yhxVaA0obYjVH/oI/ihrOoMwsLS9KSA= +github.com/Azure/azure-sdk-for-go/sdk/azcore v0.21.1/go.mod h1:fBF9PQNqB8scdgpZ3ufzaLntG0AG7C1WjPMsiFOmfHM= +github.com/Azure/azure-sdk-for-go/sdk/internal v0.8.3/go.mod h1:KLF4gFr6DcKFZwSuH8w8yEK6DpFl3LP5rhdvAb7Yz5I= +github.com/Azure/azure-sdk-for-go/sdk/internal v0.9.1 h1:sLZ/Y+P/5RRtsXWylBjB5lkgixYfm0MQPiwrSX//JSo= +github.com/Azure/azure-sdk-for-go/sdk/internal v0.9.1/go.mod h1:KLF4gFr6DcKFZwSuH8w8yEK6DpFl3LP5rhdvAb7Yz5I= +github.com/Azure/azure-sdk-for-go/sdk/storage/azblob v0.3.0 h1:Px2UA+2RvSSvv+RvJNuUB6n7rs5Wsel4dXLe90Um2n4= +github.com/Azure/azure-sdk-for-go/sdk/storage/azblob v0.3.0/go.mod h1:tPaiy8S5bQ+S5sOiDlINkp7+Ef339+Nz5L5XO+cnOHo= github.com/Azure/azure-service-bus-go v0.10.16/go.mod h1:MlkLwGGf1ewcx5jZadn0gUEty+tTg0RaElr6bPf+QhI= -github.com/Azure/azure-storage-blob-go v0.14.0 h1:1BCg74AmVdYwO3dlKwtFU1V0wU2PZdREkXvAmZJRUlM= github.com/Azure/azure-storage-blob-go v0.14.0/go.mod h1:SMqIBi+SuiQH32bvyjngEewEeXoPfKMgWlBDaYf6fck= github.com/Azure/go-amqp v0.13.0/go.mod h1:qj+o8xPCz9tMSbQ83Vp8boHahuRDl5mkNHyt1xlxUTs= github.com/Azure/go-amqp v0.13.11/go.mod h1:D5ZrjQqB1dyp1A+G73xeL/kNn7D5qHJIIsNNps7YNmk= github.com/Azure/go-amqp v0.13.12/go.mod h1:D5ZrjQqB1dyp1A+G73xeL/kNn7D5qHJIIsNNps7YNmk= -github.com/Azure/go-autorest v14.2.0+incompatible h1:V5VMDjClD3GiElqLWO7mz2MxNAK/vTfRHdAubSIPRgs= github.com/Azure/go-autorest v14.2.0+incompatible/go.mod h1:r+4oMnoxhatjLLJ6zxSWATqVooLgysK6ZNox3g/xq24= github.com/Azure/go-autorest/autorest v0.11.3/go.mod h1:JFgpikqFJ/MleTTxwepExTKnFUKKszPS8UavbQYUMuw= github.com/Azure/go-autorest/autorest v0.11.17/go.mod h1:eipySxLmqSyC5s5k1CLupqet0PSENBEDP93LQ9a8QYw= github.com/Azure/go-autorest/autorest v0.11.18/go.mod h1:dSiJPy22c3u0OtOKDNttNgqpNFY/GeWa7GH/Pz56QRA= -github.com/Azure/go-autorest/autorest v0.11.20 h1:s8H1PbCZSqg/DH7JMlOz6YMig6htWLNPsjDdlLqCx3M= github.com/Azure/go-autorest/autorest v0.11.20/go.mod h1:o3tqFY+QR40VOlk+pV4d77mORO64jOXSgEnPQgLK6JY= github.com/Azure/go-autorest/autorest/adal v0.9.0/go.mod h1:/c022QCutn2P7uY+/oQWWNcK9YU+MH96NgK+jErpbcg= github.com/Azure/go-autorest/autorest/adal v0.9.5/go.mod h1:B7KF7jKIeC9Mct5spmyCB/A8CG/sEz1vwIRGv/bbw7A= github.com/Azure/go-autorest/autorest/adal v0.9.11/go.mod h1:nBKAnTomx8gDtl+3ZCJv2v0KACFHWTB2drffI1B68Pk= github.com/Azure/go-autorest/autorest/adal v0.9.13/go.mod h1:W/MM4U6nLxnIskrw4UwWzlHfGjwUS50aOsc/I3yuU8M= github.com/Azure/go-autorest/autorest/adal v0.9.14/go.mod h1:W/MM4U6nLxnIskrw4UwWzlHfGjwUS50aOsc/I3yuU8M= -github.com/Azure/go-autorest/autorest/adal v0.9.15 h1:X+p2GF0GWyOiSmqohIaEeuNFNDY4I4EOlVuUQvFdWMk= github.com/Azure/go-autorest/autorest/adal v0.9.15/go.mod h1:tGMin8I49Yij6AQ+rvV+Xa/zwxYQB5hmsd6DkfAx2+A= github.com/Azure/go-autorest/autorest/azure/auth v0.5.8/go.mod h1:kxyKZTSfKh8OVFWPAgOgQ/frrJgeYQJPyR5fLFmXko4= github.com/Azure/go-autorest/autorest/azure/cli v0.4.2/go.mod h1:7qkJkT+j6b+hIpzMOwPChJhTqS8VbsqqgULzMNRugoM= github.com/Azure/go-autorest/autorest/azure/cli v0.4.3/go.mod h1:yAQ2b6eP/CmLPnmLvxtT1ALIY3OR1oFcCqVBi8vHiTc= -github.com/Azure/go-autorest/autorest/date v0.3.0 h1:7gUk1U5M/CQbp9WoqinNzJar+8KY+LPI6wiWrP/myHw= github.com/Azure/go-autorest/autorest/date v0.3.0/go.mod h1:BI0uouVdmngYNUzGWeSYnokU+TrmwEsOqdt8Y6sso74= github.com/Azure/go-autorest/autorest/mocks v0.4.0/go.mod h1:LTp+uSrOhSkaKrUy935gNZuuIPPVsHlr9DSOxSayd+k= github.com/Azure/go-autorest/autorest/mocks v0.4.1/go.mod h1:LTp+uSrOhSkaKrUy935gNZuuIPPVsHlr9DSOxSayd+k= github.com/Azure/go-autorest/autorest/to v0.4.0/go.mod h1:fE8iZBn7LQR7zH/9XU2NcPR4o9jEImooCeWJcYV/zLE= github.com/Azure/go-autorest/autorest/validation v0.3.1/go.mod h1:yhLgjC0Wda5DYXl6JAsWyUe4KVNffhoDhG0zVzUMo3E= github.com/Azure/go-autorest/logger v0.2.0/go.mod h1:T9E3cAhj2VqvPOtCYAvby9aBXkZmbF5NWuPV8+WeEW8= -github.com/Azure/go-autorest/logger v0.2.1 h1:IG7i4p/mDa2Ce4TRyAO8IHnVhAVF3RFU+ZtXWSmf4Tg= github.com/Azure/go-autorest/logger v0.2.1/go.mod h1:T9E3cAhj2VqvPOtCYAvby9aBXkZmbF5NWuPV8+WeEW8= -github.com/Azure/go-autorest/tracing v0.6.0 h1:TYi4+3m5t6K48TGI9AUdb+IzbnSxvnvUMfuitfgcfuo= github.com/Azure/go-autorest/tracing v0.6.0/go.mod h1:+vhtPC754Xsa23ID7GlGsrdKBpUA79WCAKPPZVC2DeU= github.com/BurntSushi/toml v0.3.1 h1:WXkYYl6Yr3qBf1K79EBnL4mak0OimBfB0XUf9Vl28OQ= github.com/BurntSushi/toml v0.3.1/go.mod h1:xHWCNGjB5oqiDr8zfno3MHue2Ht5sIBksp03qcyfWMU= @@ -218,6 +218,9 @@ github.com/devigned/tab v0.1.1/go.mod h1:XG9mPq0dFghrYvoBF3xdRrJzSTX1b7IQrvaL9mz github.com/dgrijalva/jwt-go v3.2.0+incompatible/go.mod h1:E3ru+11k8xSBh+hMPgOLZmtrrCbhqsmaPHjLKYnJCaQ= github.com/dimchansky/utfbom v1.1.0/go.mod h1:rO41eb7gLfo8SF1jd9F8HplJm1Fewwi4mQvIirEdv+8= github.com/dimchansky/utfbom v1.1.1/go.mod h1:SxdoEBH5qIqFocHMyGOXVAybYJdr71b1Q/j0mACtrfE= +github.com/dnaeon/go-vcr v1.1.0/go.mod h1:M7tiix8f0r6mKKJ3Yq/kqU1OYf3MnfmBWVbPx/yU9ko= +github.com/dnaeon/go-vcr v1.2.0 h1:zHCHvJYTMh1N7xnV7zf1m1GPBF9Ad0Jk/whtQ1663qI= +github.com/dnaeon/go-vcr v1.2.0/go.mod h1:R4UdLID7HZT3taECzJs4YgbbH6PIGXB6W/sc5OLb6RQ= github.com/drakkan/crypto v0.0.0-20220215181150-74469fa99b22 h1:yHFyJbCfvTY65bTyPOMRGqplA5GNfdZhBVOiYGNeCtY= github.com/drakkan/crypto v0.0.0-20220215181150-74469fa99b22/go.mod h1:SiM6ypd8Xu1xldObYtbDztuUU7xUzMnUULfphXFZmro= github.com/drakkan/ftp v0.0.0-20201114075148-9b9adce499a9 h1:LPH1dEblAOO/LoG7yHPMtBLXhQmjaga91/DDjWk9jWA= @@ -295,7 +298,6 @@ github.com/gofrs/flock v0.8.1/go.mod h1:F1TvTiK9OcQqauNUHlbJvyl9Qa1QvF/gOUDKA14j github.com/gofrs/uuid v3.2.0+incompatible/go.mod h1:b2aQJv3Z4Fp6yNu3cdSllBxTCLRxnplIgP/c0N/04lM= github.com/gogo/protobuf v1.1.1/go.mod h1:r8qH/GZQm5c6nD/R0oafs1akxWv10x8SbQlK7atdtwQ= github.com/gogo/protobuf v1.3.2/go.mod h1:P1XiOD3dCwIKUDQYPy72D8LYyHL2YPYrpS2s69NZV8Q= -github.com/golang-jwt/jwt/v4 v4.0.0 h1:RAqyYixv1p7uEnocuy8P1nru5wprCh/MH2BIlW5z5/o= github.com/golang-jwt/jwt/v4 v4.0.0/go.mod h1:/xlHOz8bRuivTWchD4jCa+NbatV+wEUSzwAxVc6locg= github.com/golang-sql/civil v0.0.0-20190719163853-cb61b32ac6fe/go.mod h1:8vg3r2VgvsThLBIFL93Qb5yWzgyZWhEmBwUJWevAkK0= github.com/golang/glog v0.0.0-20160126235308-23def4e6c14b/go.mod h1:SBH7ygxi8pfUlaOkMMuAQtPIUF8ecWP5IEl/CR7VP2Q= @@ -569,8 +571,6 @@ github.com/mattn/go-colorable v0.1.9/go.mod h1:u6P/XSegPjTcexA+o6vUJrdnUu04hMope github.com/mattn/go-colorable v0.1.12 h1:jF+Du6AlPIjs2BiUiQlKOX0rt3SujHxPnksPKZbaA40= github.com/mattn/go-colorable v0.1.12/go.mod h1:u5H1YNBxpqRaxsYJYSkiCWKzEfiAb1Gb520KVy5xxl4= github.com/mattn/go-ieproxy v0.0.1/go.mod h1:pYabZ6IHcRpFh7vIaLfK7rdcWgFEb3SFJ6/gNWuh88E= -github.com/mattn/go-ieproxy v0.0.3 h1:YkaHmK1CzE5C4O7A3hv3TCbfNDPSCf0RKZFX+VhBeYk= -github.com/mattn/go-ieproxy v0.0.3/go.mod h1:6ZpRmhBaYuBX1U2za+9rC9iCGLsSp2tftelZne7CPko= github.com/mattn/go-isatty v0.0.3/go.mod h1:M+lRXTBqGeGNdLjl/ufCoiOlB5xdOkqRJdNxMWT7Zi4= github.com/mattn/go-isatty v0.0.5/go.mod h1:Iq45c/XA43vh69/j3iqttzPXn0bhXyGjM0Hdxcsrc5s= github.com/mattn/go-isatty v0.0.7/go.mod h1:Iq45c/XA43vh69/j3iqttzPXn0bhXyGjM0Hdxcsrc5s= @@ -615,6 +615,7 @@ github.com/modern-go/concurrent v0.0.0-20180306012644-bacd9c7ef1dd/go.mod h1:6dJ github.com/modern-go/reflect2 v0.0.0-20180701023420-4b7aa43c6742/go.mod h1:bx2lNnkwVCuqBIxFjflWJWanXIb3RllmbCylyMrvgv0= github.com/modern-go/reflect2 v1.0.1/go.mod h1:bx2lNnkwVCuqBIxFjflWJWanXIb3RllmbCylyMrvgv0= github.com/modern-go/reflect2 v1.0.2/go.mod h1:yWuevngMOJpCy52FWWMvUC8ws7m/LJsjYzDa0/r8luk= +github.com/modocache/gover v0.0.0-20171022184752-b58185e213c5/go.mod h1:caMODM3PzxT8aQXRPkAt8xlV/e7d7w8GM5g0fa5F0D8= github.com/mwitkow/go-conntrack v0.0.0-20161129095857-cc309e4a2223/go.mod h1:qRWi+5nqEBWmkhHvq77mSJWrCKwh8bxhgT7d/eI7P4U= github.com/mwitkow/go-conntrack v0.0.0-20190716064945-2f068394615f/go.mod h1:qRWi+5nqEBWmkhHvq77mSJWrCKwh8bxhgT7d/eI7P4U= github.com/oklog/run v1.0.0/go.mod h1:dlhp/R75TPv97u0XWUtDeV/lRKWPKSdTuV0TZvrmrQA= @@ -949,7 +950,6 @@ golang.org/x/sys v0.0.0-20211124211545-fe61309f8881/go.mod h1:oPkhp1MJrh7nUepCBc golang.org/x/sys v0.0.0-20211205182925-97ca703d548d/go.mod h1:oPkhp1MJrh7nUepCBck5+mAzfO9JrbApNNgaTdGDITg= golang.org/x/sys v0.0.0-20211210111614-af8b64212486/go.mod h1:oPkhp1MJrh7nUepCBck5+mAzfO9JrbApNNgaTdGDITg= golang.org/x/sys v0.0.0-20211216021012-1d35b9e2eb4e/go.mod h1:oPkhp1MJrh7nUepCBck5+mAzfO9JrbApNNgaTdGDITg= -golang.org/x/sys v0.0.0-20220110181412-a018aaa089fe/go.mod h1:oPkhp1MJrh7nUepCBck5+mAzfO9JrbApNNgaTdGDITg= golang.org/x/sys v0.0.0-20220111092808-5a964db01320/go.mod h1:oPkhp1MJrh7nUepCBck5+mAzfO9JrbApNNgaTdGDITg= golang.org/x/sys v0.0.0-20220114195835-da31bd327af9/go.mod h1:oPkhp1MJrh7nUepCBck5+mAzfO9JrbApNNgaTdGDITg= golang.org/x/sys v0.0.0-20220128215802-99c3d69c2c27/go.mod h1:oPkhp1MJrh7nUepCBck5+mAzfO9JrbApNNgaTdGDITg= diff --git a/util/util.go b/util/util.go index db78ce67..ede04f40 100644 --- a/util/util.go +++ b/util/util.go @@ -145,12 +145,36 @@ func GetIPFromRemoteAddress(remoteAddress string) string { // NilIfEmpty returns nil if the input string is empty func NilIfEmpty(s string) *string { - if len(s) == 0 { + if s == "" { return nil } return &s } +// GetStringFromPointer returns the string value or empty if nil +func GetStringFromPointer(val *string) string { + if val == nil { + return "" + } + return *val +} + +// GetIntFromPointer returns the int value or zero +func GetIntFromPointer(val *int64) int64 { + if val == nil { + return 0 + } + return *val +} + +// GetTimeFromPointer returns the time value or now +func GetTimeFromPointer(val *time.Time) time.Time { + if val == nil { + return time.Now() + } + return *val +} + // GenerateRSAKeys generate rsa private and public keys and write the // private key to specified file and the public key to the specified // file adding the .pub suffix diff --git a/vfs/azblobfs.go b/vfs/azblobfs.go index 08fd5b95..6f8c17f5 100644 --- a/vfs/azblobfs.go +++ b/vfs/azblobfs.go @@ -12,7 +12,6 @@ import ( "io" "mime" "net/http" - "net/url" "os" "path" "path/filepath" @@ -20,7 +19,8 @@ import ( "sync" "time" - "github.com/Azure/azure-storage-blob-go/azblob" + "github.com/Azure/azure-sdk-for-go/sdk/azcore/policy" + "github.com/Azure/azure-sdk-for-go/sdk/storage/azblob" "github.com/eikenb/pipeat" "github.com/pkg/sftp" @@ -31,23 +31,22 @@ import ( "github.com/drakkan/sftpgo/v2/version" ) -const azureDefaultEndpoint = "blob.core.windows.net" - -// max time of an azure web request response window (whether or not data is flowing) -// this is the same value used in rclone -var maxTryTimeout = time.Hour * 24 * 365 +const ( + azureDefaultEndpoint = "blob.core.windows.net" + maxResultsPerPage = 1000 +) // AzureBlobFs is a Fs implementation for Azure Blob storage. type AzureBlobFs struct { connectionID string localTempDir string // if not empty this fs is mouted as virtual folder in the specified path - mountPath string - config *AzBlobFsConfig - svc *azblob.ServiceURL - containerURL azblob.ContainerURL - ctxTimeout time.Duration - ctxLongTimeout time.Duration + mountPath string + config *AzBlobFsConfig + hasContainerAccess bool + containerClient azblob.ContainerClient + ctxTimeout time.Duration + ctxLongTimeout time.Duration } func init() { @@ -69,7 +68,7 @@ func NewAzBlobFs(connectionID, localTempDir, mountPath string, config AzBlobFsCo mountPath: mountPath, config: &config, ctxTimeout: 30 * time.Second, - ctxLongTimeout: 300 * time.Second, + ctxLongTimeout: 90 * time.Second, } if err := fs.config.Validate(); err != nil { return fs, err @@ -84,39 +83,31 @@ func NewAzBlobFs(connectionID, localTempDir, mountPath string, config AzBlobFsCo fs.setConfigDefaults() version := version.Get() - telemetryValue := fmt.Sprintf("SFTPGo-%v_%v", version.Version, version.CommitHash) + clientOptions := &azblob.ClientOptions{ + Telemetry: policy.TelemetryOptions{ + ApplicationID: fmt.Sprintf("SFTPGo-%v_%v", version.Version, version.CommitHash), + }, + } if fs.config.SASURL.GetPayload() != "" { - u, err := url.Parse(fs.config.SASURL.GetPayload()) - if err != nil { - return fs, fmt.Errorf("invalid credentials: %v", err) - } - pipeline := azblob.NewPipeline(azblob.NewAnonymousCredential(), azblob.PipelineOptions{ - Retry: azblob.RetryOptions{ - TryTimeout: maxTryTimeout, - }, - Telemetry: azblob.TelemetryOptions{ - Value: telemetryValue, - }, - }) - // Check if we have container level SAS or account level SAS - parts := azblob.NewBlobURLParts(*u) + parts := azblob.NewBlobURLParts(fs.config.SASURL.GetPayload()) if parts.ContainerName != "" { if fs.config.Container != "" && fs.config.Container != parts.ContainerName { return fs, fmt.Errorf("container name in SAS URL %#v and container provided %#v do not match", parts.ContainerName, fs.config.Container) } fs.config.Container = parts.ContainerName - fs.svc = nil - fs.containerURL = azblob.NewContainerURL(*u, pipeline) } else { if fs.config.Container == "" { return fs, errors.New("container is required with this SAS URL") } - serviceURL := azblob.NewServiceURL(*u, pipeline) - fs.svc = &serviceURL - fs.containerURL = fs.svc.NewContainerURL(fs.config.Container) } + svc, err := azblob.NewServiceClientWithNoCredential(fs.config.SASURL.GetPayload(), clientOptions) + if err != nil { + return fs, fmt.Errorf("invalid credentials: %v", err) + } + fs.hasContainerAccess = false + fs.containerClient = svc.NewContainerClient(fs.config.Container) return fs, nil } @@ -124,28 +115,18 @@ func NewAzBlobFs(connectionID, localTempDir, mountPath string, config AzBlobFsCo if err != nil { return fs, fmt.Errorf("invalid credentials: %v", err) } - var u *url.URL + var endpoint string if fs.config.UseEmulator { - // for the emulator we expect the endpoint prefixed with the protocol, for example: - // http://127.0.0.1:10000 - u, err = url.Parse(fmt.Sprintf("%s/%s", fs.config.Endpoint, fs.config.AccountName)) + endpoint = fmt.Sprintf("%s/%s", fs.config.Endpoint, fs.config.AccountName) } else { - u, err = url.Parse(fmt.Sprintf("https://%s.%s", fs.config.AccountName, fs.config.Endpoint)) + endpoint = fmt.Sprintf("https://%s.%s/", fs.config.AccountName, fs.config.Endpoint) } + svc, err := azblob.NewServiceClientWithSharedKey(endpoint, credential, clientOptions) if err != nil { return fs, fmt.Errorf("invalid credentials: %v", err) } - pipeline := azblob.NewPipeline(credential, azblob.PipelineOptions{ - Retry: azblob.RetryOptions{ - TryTimeout: maxTryTimeout, - }, - Telemetry: azblob.TelemetryOptions{ - Value: telemetryValue, - }, - }) - serviceURL := azblob.NewServiceURL(*u, pipeline) - fs.svc = &serviceURL - fs.containerURL = fs.svc.NewContainerURL(fs.config.Container) + fs.hasContainerAccess = true + fs.containerClient = svc.NewContainerClient(fs.config.Container) return fs, nil } @@ -165,7 +146,7 @@ func (fs *AzureBlobFs) ConnectionID() string { // Stat returns a FileInfo describing the named file func (fs *AzureBlobFs) Stat(name string) (os.FileInfo, error) { if name == "" || name == "." { - if fs.svc != nil { + if fs.hasContainerAccess { err := fs.checkIfBucketExists() if err != nil { return nil, err @@ -179,10 +160,12 @@ func (fs *AzureBlobFs) Stat(name string) (os.FileInfo, error) { attrs, err := fs.headObject(name) if err == nil { - isDir := (attrs.ContentType() == dirMimeType) + contentType := util.GetStringFromPointer(attrs.ContentType) + isDir := contentType == dirMimeType metric.AZListObjectsCompleted(nil) - return updateFileInfoModTime(fs.getStorageID(), name, NewFileInfo(name, isDir, attrs.ContentLength(), - attrs.LastModified(), false)) + return updateFileInfoModTime(fs.getStorageID(), name, NewFileInfo(name, isDir, + util.GetIntFromPointer(attrs.ContentLength), + util.GetTimeFromPointer(attrs.LastModified), false)) } if !fs.IsNotExist(err) { return nil, err @@ -209,24 +192,34 @@ func (fs *AzureBlobFs) Open(name string, offset int64) (File, *pipeat.PipeReader if err != nil { return nil, nil, nil, err } - blobBlockURL := fs.containerURL.NewBlockBlobURL(name) ctx, cancelFn := context.WithCancel(context.Background()) - blobDownloadResponse, err := blobBlockURL.Download(ctx, offset, azblob.CountToEnd, azblob.BlobAccessConditions{}, false, - azblob.ClientProvidedKeyOptions{}) + + blockBlob := fs.containerClient.NewBlockBlobClient(name) + blobDownloadResponse, err := blockBlob.Download(ctx, &azblob.DownloadBlobOptions{ + Offset: &offset, + }) if err != nil { r.Close() w.Close() cancelFn() return nil, nil, nil, err } - body := blobDownloadResponse.Body(azblob.RetryReaderOptions{ - MaxRetryRequests: 3, + body := blobDownloadResponse.Body(&azblob.RetryReaderOptions{ + MaxRetryRequests: 2, }) go func() { defer cancelFn() defer body.Close() + /*err := blockBlob.DownloadBlobToWriterAt(ctx, offset, 0, w, azblob.HighLevelDownloadFromBlobOptions{ + // add download part size and concurrency + BlockSize: fs.config.UploadPartSize, + Parallelism: uint16(fs.config.UploadConcurrency), + RetryReaderOptionsPerBlock: azblob.RetryReaderOptions{ + MaxRetryRequests: 2, + }, + })*/ n, err := io.Copy(w, body) w.CloseWithError(err) //nolint:errcheck fsLog(fs, logger.LevelDebug, "download completed, path: %#v size: %v, err: %v", name, n, err) @@ -242,10 +235,10 @@ func (fs *AzureBlobFs) Create(name string, flag int) (File, *PipeWriter, func(), if err != nil { return nil, nil, nil, err } - p := NewPipeWriter(w) - blobBlockURL := fs.containerURL.NewBlockBlobURL(name) ctx, cancelFn := context.WithCancel(context.Background()) + p := NewPipeWriter(w) + blockBlob := fs.containerClient.NewBlockBlobClient(name) headers := azblob.BlobHTTPHeaders{} var contentType string if flag == -1 { @@ -254,22 +247,22 @@ func (fs *AzureBlobFs) Create(name string, flag int) (File, *PipeWriter, func(), contentType = mime.TypeByExtension(path.Ext(name)) } if contentType != "" { - headers.ContentType = contentType + headers.BlobContentType = &contentType } go func() { defer cancelFn() /*uploadOptions := azblob.UploadStreamToBlockBlobOptions{ - BufferSize: int(fs.config.UploadPartSize), - BlobHTTPHeaders: headers, - MaxBuffers: fs.config.UploadConcurrency, + BufferSize: int(fs.config.UploadPartSize), + MaxBuffers: fs.config.UploadConcurrency, + HTTPHeaders: &headers, } - // UploadStreamToBlockBlob seems to have issues if there is an error, for example - // if we shutdown Azurite while uploading it hangs, so we use our own wrapper for - // the low level functions - _, err := azblob.UploadStreamToBlockBlob(ctx, r, blobBlockURL, uploadOptions)*/ - err := fs.handleMultipartUpload(ctx, r, &blobBlockURL, &headers) + if fs.config.AccessTier != "" { + uploadOptions.AccessTier = (*azblob.AccessTier)(&fs.config.AccessTier) + } + _, err := blockBlob.UploadStreamToBlockBlob(ctx, r, uploadOptions)*/ + err := fs.handleMultipartUpload(ctx, r, blockBlob, &headers) r.CloseWithError(err) //nolint:errcheck p.Done(err) fsLog(fs, logger.LevelDebug, "upload completed, path: %#v, readed bytes: %v, err: %v", name, r.GetReadedBytes(), err) @@ -301,26 +294,25 @@ func (fs *AzureBlobFs) Rename(source, target string) error { return fmt.Errorf("cannot rename non empty directory: %#v", source) } } - dstBlobURL := fs.containerURL.NewBlobURL(target) - srcURL := fs.containerURL.NewBlobURL(source).URL() + dstBlob := fs.containerClient.NewBlockBlobClient(target) + srcURL := fs.containerClient.NewBlockBlobClient(source).URL() - md := azblob.Metadata{} - mac := azblob.ModifiedAccessConditions{} - bac := azblob.BlobAccessConditions{} - ctx, cancelFn := context.WithDeadline(context.Background(), time.Now().Add(fs.ctxTimeout)) + ctx, cancelFn := context.WithDeadline(context.Background(), time.Now().Add(fs.ctxLongTimeout)) defer cancelFn() - resp, err := dstBlobURL.StartCopyFromURL(ctx, srcURL, md, mac, bac, azblob.AccessTierType(fs.config.AccessTier), nil) + resp, err := dstBlob.StartCopyFromURL(ctx, srcURL, fs.getCopyOptions()) if err != nil { metric.AZCopyObjectCompleted(err) return err } - copyStatus := resp.CopyStatus() + copyStatus := azblob.CopyStatusType(util.GetStringFromPointer((*string)(resp.CopyStatus))) nErrors := 0 - for copyStatus == azblob.CopyStatusPending { + for copyStatus == azblob.CopyStatusTypePending { // Poll until the copy is complete. time.Sleep(500 * time.Millisecond) - propertiesResp, err := dstBlobURL.GetProperties(ctx, azblob.BlobAccessConditions{}, azblob.ClientProvidedKeyOptions{}) + resp, err := dstBlob.GetProperties(ctx, &azblob.GetBlobPropertiesOptions{ + BlobAccessConditions: &azblob.BlobAccessConditions{}, + }) if err != nil { // A GetProperties failure may be transient, so allow a couple // of them before giving up. @@ -330,14 +322,15 @@ func (fs *AzureBlobFs) Rename(source, target string) error { return err } } else { - copyStatus = propertiesResp.CopyStatus() + copyStatus = azblob.CopyStatusType(util.GetStringFromPointer((*string)(resp.CopyStatus))) } } - if copyStatus != azblob.CopyStatusSuccess { + if copyStatus != azblob.CopyStatusTypeSuccess { err := fmt.Errorf("copy failed with status: %s", copyStatus) metric.AZCopyObjectCompleted(err) return err } + metric.AZCopyObjectCompleted(nil) if plugin.Handler.HasMetadater() { if !fi.IsDir() { @@ -363,11 +356,13 @@ func (fs *AzureBlobFs) Remove(name string, isDir bool) error { return fmt.Errorf("cannot remove non empty directory: %#v", name) } } - blobBlockURL := fs.containerURL.NewBlockBlobURL(name) + blobBlock := fs.containerClient.NewBlockBlobClient(name) ctx, cancelFn := context.WithDeadline(context.Background(), time.Now().Add(fs.ctxTimeout)) defer cancelFn() - _, err := blobBlockURL.Delete(ctx, azblob.DeleteSnapshotsOptionNone, azblob.BlobAccessConditions{}) + _, err := blobBlock.Delete(ctx, &azblob.DeleteBlobOptions{ + DeleteSnapshots: azblob.DeleteSnapshotsOptionTypeInclude.ToPtr(), + }) metric.AZDeleteObjectCompleted(err) if plugin.Handler.HasMetadater() && err == nil && !isDir { if errMetadata := plugin.Handler.RemoveMetadata(fs.getStorageID(), ensureAbsPath(name)); errMetadata != nil { @@ -454,66 +449,67 @@ func (fs *AzureBlobFs) ReadDir(dirname string) ([]os.FileInfo, error) { } prefixes := make(map[string]bool) - for marker := (azblob.Marker{}); marker.NotDone(); { + maxResults := int32(maxResultsPerPage) + pager := fs.containerClient.ListBlobsHierarchy("/", &azblob.ContainerListBlobHierarchySegmentOptions{ + Include: []azblob.ListBlobsIncludeItem{}, + Prefix: &prefix, + Maxresults: &maxResults, + }) + + hasNext := true + for hasNext { ctx, cancelFn := context.WithDeadline(context.Background(), time.Now().Add(fs.ctxTimeout)) defer cancelFn() - listBlob, err := fs.containerURL.ListBlobsHierarchySegment(ctx, marker, "/", azblob.ListBlobsSegmentOptions{ - Details: azblob.BlobListingDetails{ - Copy: false, - Metadata: false, - Snapshots: false, - UncommittedBlobs: false, - Deleted: false, - }, - Prefix: prefix, - }) - if err != nil { - metric.AZListObjectsCompleted(err) - return nil, err - } - marker = listBlob.NextMarker - for _, blobPrefix := range listBlob.Segment.BlobPrefixes { - // we don't support prefixes == "/" this will be sent if a key starts with "/" - if blobPrefix.Name == "/" { - continue - } - // sometime we have duplicate prefixes, maybe an Azurite bug - name := strings.TrimPrefix(blobPrefix.Name, prefix) - if _, ok := prefixes[strings.TrimSuffix(name, "/")]; ok { - continue - } - result = append(result, NewFileInfo(name, true, 0, time.Now(), false)) - prefixes[strings.TrimSuffix(name, "/")] = true - } - for idx := range listBlob.Segment.BlobItems { - blobInfo := &listBlob.Segment.BlobItems[idx] - name := strings.TrimPrefix(blobInfo.Name, prefix) - size := int64(0) - if blobInfo.Properties.ContentLength != nil { - size = *blobInfo.Properties.ContentLength - } - isDir := false - if blobInfo.Properties.ContentType != nil { - isDir = (*blobInfo.Properties.ContentType == dirMimeType) - if isDir { - // check if the dir is already included, it will be sent as blob prefix if it contains at least one item - if _, ok := prefixes[name]; ok { - continue - } - prefixes[name] = true + if hasNext = pager.NextPage(ctx); hasNext { + resp := pager.PageResponse() + + for _, blobPrefix := range resp.ContainerListBlobHierarchySegmentResult.Segment.BlobPrefixes { + name := util.GetStringFromPointer(blobPrefix.Name) + // we don't support prefixes == "/" this will be sent if a key starts with "/" + if name == "" || name == "/" { + continue } + // sometime we have duplicate prefixes, maybe an Azurite bug + name = strings.TrimPrefix(name, prefix) + if _, ok := prefixes[strings.TrimSuffix(name, "/")]; ok { + continue + } + result = append(result, NewFileInfo(name, true, 0, time.Now(), false)) + prefixes[strings.TrimSuffix(name, "/")] = true } - modTime := blobInfo.Properties.LastModified - if t, ok := modTimes[name]; ok { - modTime = util.GetTimeFromMsecSinceEpoch(t) + + for _, blobItem := range resp.ContainerListBlobHierarchySegmentResult.Segment.BlobItems { + name := util.GetStringFromPointer(blobItem.Name) + name = strings.TrimPrefix(name, prefix) + size := int64(0) + isDir := false + modTime := time.Now() + if blobItem.Properties != nil { + size = util.GetIntFromPointer(blobItem.Properties.ContentLength) + modTime = util.GetTimeFromPointer(blobItem.Properties.LastModified) + contentType := util.GetStringFromPointer(blobItem.Properties.ContentType) + isDir = (contentType == dirMimeType) + if isDir { + // check if the dir is already included, it will be sent as blob prefix if it contains at least one item + if _, ok := prefixes[name]; ok { + continue + } + prefixes[name] = true + } + } + if t, ok := modTimes[name]; ok { + modTime = util.GetTimeFromMsecSinceEpoch(t) + } + result = append(result, NewFileInfo(name, isDir, size, modTime, false)) } - result = append(result, NewFileInfo(name, isDir, size, modTime, false)) } } - metric.AZListObjectsCompleted(nil) - return result, nil + err = pager.Err() + metric.AZListObjectsCompleted(err) + + return result, err } // IsUploadResumeSupported returns true if resuming uploads is supported. @@ -535,15 +531,9 @@ func (*AzureBlobFs) IsNotExist(err error) bool { if err == nil { return false } - - if storageErr, ok := err.(azblob.StorageError); ok { - if storageErr.Response().StatusCode == http.StatusNotFound { //nolint:bodyclose - return true - } - if storageErr.ServiceCode() == azblob.ServiceCodeContainerNotFound || - storageErr.ServiceCode() == azblob.ServiceCodeBlobNotFound { - return true - } + var errResp *azblob.StorageError + if errors.As(err, &errResp) { + return errResp.StatusCode() == http.StatusNotFound } return strings.Contains(err.Error(), "404") @@ -555,17 +545,11 @@ func (*AzureBlobFs) IsPermission(err error) bool { if err == nil { return false } - if storageErr, ok := err.(azblob.StorageError); ok { - code := storageErr.Response().StatusCode //nolint:bodyclose - if code == http.StatusForbidden || code == http.StatusUnauthorized { - return true - } - if storageErr.ServiceCode() == azblob.ServiceCodeInsufficientAccountPermissions || - storageErr.ServiceCode() == azblob.ServiceCodeInvalidAuthenticationInfo || - storageErr.ServiceCode() == azblob.ServiceCodeUnauthorizedBlobOverwrite { - return true - } + var errResp *azblob.StorageError + if errors.As(err, &errResp) { + return errResp.StatusCode() == http.StatusForbidden } + return strings.Contains(err.Error(), "403") } @@ -590,45 +574,38 @@ func (fs *AzureBlobFs) ScanRootDirContents() (int, int64, error) { numFiles := 0 size := int64(0) - for marker := (azblob.Marker{}); marker.NotDone(); { + maxResults := int32(maxResultsPerPage) + pager := fs.containerClient.ListBlobsFlat(&azblob.ContainerListBlobFlatSegmentOptions{ + Prefix: &fs.config.KeyPrefix, + Maxresults: &maxResults, + }) + + hasNext := true + for hasNext { ctx, cancelFn := context.WithDeadline(context.Background(), time.Now().Add(fs.ctxTimeout)) defer cancelFn() - listBlob, err := fs.containerURL.ListBlobsFlatSegment(ctx, marker, azblob.ListBlobsSegmentOptions{ - Details: azblob.BlobListingDetails{ - Copy: false, - Metadata: false, - Snapshots: false, - UncommittedBlobs: false, - Deleted: false, - }, - Prefix: fs.config.KeyPrefix, - }) - if err != nil { - metric.AZListObjectsCompleted(err) - return numFiles, size, err - } - marker = listBlob.NextMarker - for idx := range listBlob.Segment.BlobItems { - blobInfo := &listBlob.Segment.BlobItems[idx] - isDir := false - if blobInfo.Properties.ContentType != nil { - isDir = (*blobInfo.Properties.ContentType == dirMimeType) + if hasNext = pager.NextPage(ctx); hasNext { + resp := pager.PageResponse() + for _, blobItem := range resp.ContainerListBlobFlatSegmentResult.Segment.BlobItems { + if blobItem.Properties != nil { + contentType := util.GetStringFromPointer(blobItem.Properties.ContentType) + isDir := (contentType == dirMimeType) + blobSize := util.GetIntFromPointer(blobItem.Properties.ContentLength) + if isDir && blobSize == 0 { + continue + } + numFiles++ + size += blobSize + } } - blobSize := int64(0) - if blobInfo.Properties.ContentLength != nil { - blobSize = *blobInfo.Properties.ContentLength - } - if isDir && blobSize == 0 { - continue - } - numFiles++ - size += blobSize } } - metric.AZListObjectsCompleted(nil) - return numFiles, size, nil + err := pager.Err() + metric.AZListObjectsCompleted(err) + + return numFiles, size, err } func (fs *AzureBlobFs) getFileNamesInPrefix(fsPrefix string) (map[string]bool, error) { @@ -638,40 +615,39 @@ func (fs *AzureBlobFs) getFileNamesInPrefix(fsPrefix string) (map[string]bool, e prefix = strings.TrimPrefix(fsPrefix, "/") } - for marker := (azblob.Marker{}); marker.NotDone(); { + maxResults := int32(maxResultsPerPage) + pager := fs.containerClient.ListBlobsHierarchy("/", &azblob.ContainerListBlobHierarchySegmentOptions{ + Include: []azblob.ListBlobsIncludeItem{}, + Prefix: &prefix, + Maxresults: &maxResults, + }) + + hasNext := true + for hasNext { ctx, cancelFn := context.WithDeadline(context.Background(), time.Now().Add(fs.ctxTimeout)) defer cancelFn() - listBlob, err := fs.containerURL.ListBlobsHierarchySegment(ctx, marker, "/", azblob.ListBlobsSegmentOptions{ - Details: azblob.BlobListingDetails{ - Copy: false, - Metadata: false, - Snapshots: false, - UncommittedBlobs: false, - Deleted: false, - }, - Prefix: prefix, - }) - if err != nil { - metric.AZListObjectsCompleted(err) - return fileNames, err - } - marker = listBlob.NextMarker - for idx := range listBlob.Segment.BlobItems { - blobInfo := &listBlob.Segment.BlobItems[idx] - name := strings.TrimPrefix(blobInfo.Name, prefix) - if blobInfo.Properties.ContentType != nil { - if *blobInfo.Properties.ContentType == dirMimeType { - continue + if hasNext = pager.NextPage(ctx); hasNext { + resp := pager.PageResponse() + for _, blobItem := range resp.ContainerListBlobHierarchySegmentResult.Segment.BlobItems { + name := util.GetStringFromPointer(blobItem.Name) + name = strings.TrimPrefix(name, prefix) + if blobItem.Properties != nil { + contentType := util.GetStringFromPointer(blobItem.Properties.ContentType) + isDir := (contentType == dirMimeType) + if isDir { + continue + } + fileNames[name] = true } } - - fileNames[name] = true } } - metric.AZListObjectsCompleted(nil) - return fileNames, nil + err := pager.Err() + metric.AZListObjectsCompleted(err) + + return fileNames, err } // CheckMetadata checks the metadata consistency @@ -723,45 +699,48 @@ func (fs *AzureBlobFs) Walk(root string, walkFn filepath.WalkFunc) error { prefix += "/" } } - for marker := (azblob.Marker{}); marker.NotDone(); { + + maxResults := int32(maxResultsPerPage) + pager := fs.containerClient.ListBlobsFlat(&azblob.ContainerListBlobFlatSegmentOptions{ + Prefix: &prefix, + Maxresults: &maxResults, + }) + + hasNext := true + for hasNext { ctx, cancelFn := context.WithDeadline(context.Background(), time.Now().Add(fs.ctxTimeout)) defer cancelFn() - listBlob, err := fs.containerURL.ListBlobsFlatSegment(ctx, marker, azblob.ListBlobsSegmentOptions{ - Details: azblob.BlobListingDetails{ - Copy: false, - Metadata: false, - Snapshots: false, - UncommittedBlobs: false, - Deleted: false, - }, - Prefix: prefix, - }) - if err != nil { - metric.AZListObjectsCompleted(err) - return err - } - marker = listBlob.NextMarker - for idx := range listBlob.Segment.BlobItems { - blobInfo := &listBlob.Segment.BlobItems[idx] - isDir := false - if blobInfo.Properties.ContentType != nil { - isDir = (*blobInfo.Properties.ContentType == dirMimeType) - } - if fs.isEqual(blobInfo.Name, prefix) { - continue - } - blobSize := int64(0) - if blobInfo.Properties.ContentLength != nil { - blobSize = *blobInfo.Properties.ContentLength - } - err = walkFn(blobInfo.Name, NewFileInfo(blobInfo.Name, isDir, blobSize, blobInfo.Properties.LastModified, false), nil) - if err != nil { - return err + if hasNext = pager.NextPage(ctx); hasNext { + resp := pager.PageResponse() + for _, blobItem := range resp.ContainerListBlobFlatSegmentResult.Segment.BlobItems { + name := util.GetStringFromPointer(blobItem.Name) + if fs.isEqual(name, prefix) { + continue + } + blobSize := int64(0) + lastModified := time.Now() + isDir := false + if blobItem.Properties != nil { + contentType := util.GetStringFromPointer(blobItem.Properties.ContentType) + isDir = (contentType == dirMimeType) + blobSize = util.GetIntFromPointer(blobItem.Properties.ContentLength) + lastModified = util.GetTimeFromPointer(blobItem.Properties.LastModified) + } + err := walkFn(name, NewFileInfo(name, isDir, blobSize, lastModified, false), nil) + if err != nil { + return err + } } } } + err := pager.Err() + if err != nil { + metric.AZListObjectsCompleted(err) + return err + } + metric.AZListObjectsCompleted(nil) return walkFn(root, NewFileInfo(root, true, 0, time.Now(), false), nil) } @@ -787,14 +766,16 @@ func (fs *AzureBlobFs) ResolvePath(virtualPath string) (string, error) { return fs.Join(fs.config.KeyPrefix, strings.TrimPrefix(virtualPath, "/")), nil } -func (fs *AzureBlobFs) headObject(name string) (*azblob.BlobGetPropertiesResponse, error) { +func (fs *AzureBlobFs) headObject(name string) (azblob.GetBlobPropertiesResponse, error) { ctx, cancelFn := context.WithDeadline(context.Background(), time.Now().Add(fs.ctxTimeout)) defer cancelFn() - blobBlockURL := fs.containerURL.NewBlockBlobURL(name) - response, err := blobBlockURL.GetProperties(ctx, azblob.BlobAccessConditions{}, azblob.ClientProvidedKeyOptions{}) + blobClient := fs.containerClient.NewBlockBlobClient(name) + resp, err := blobClient.GetProperties(ctx, &azblob.GetBlobPropertiesOptions{ + BlobAccessConditions: &azblob.BlobAccessConditions{}, + }) metric.AZHeadObjectCompleted(err) - return response, err + return resp, err } // GetMimeType returns the content type @@ -803,7 +784,7 @@ func (fs *AzureBlobFs) GetMimeType(name string) (string, error) { if err != nil { return "", err } - return response.ContentType(), nil + return util.GetStringFromPointer(response.ContentType), nil } // Close closes the fs @@ -851,16 +832,13 @@ func (fs *AzureBlobFs) setConfigDefaults() { if fs.config.UploadConcurrency == 0 { fs.config.UploadConcurrency = 2 } - if fs.config.AccessTier == "" { - fs.config.AccessTier = string(azblob.AccessTierNone) - } } func (fs *AzureBlobFs) checkIfBucketExists() error { ctx, cancelFn := context.WithDeadline(context.Background(), time.Now().Add(fs.ctxTimeout)) defer cancelFn() - _, err := fs.containerURL.GetProperties(ctx, azblob.LeaseAccessConditions{}) + _, err := fs.containerClient.GetProperties(ctx, &azblob.GetPropertiesOptionsContainer{}) metric.AZHeadContainerCompleted(err) return err } @@ -874,30 +852,32 @@ func (fs *AzureBlobFs) hasContents(name string) (bool, error) { prefix += "/" } } + + maxResults := int32(1) + pager := fs.containerClient.ListBlobsFlat(&azblob.ContainerListBlobFlatSegmentOptions{ + Maxresults: &maxResults, + Prefix: &prefix, + }) + ctx, cancelFn := context.WithDeadline(context.Background(), time.Now().Add(fs.ctxTimeout)) defer cancelFn() - listBlob, err := fs.containerURL.ListBlobsFlatSegment(ctx, azblob.Marker{}, azblob.ListBlobsSegmentOptions{ - Details: azblob.BlobListingDetails{ - Copy: false, - Metadata: false, - Snapshots: false, - UncommittedBlobs: false, - Deleted: false, - }, - Prefix: prefix, - MaxResults: 1, - }) - metric.AZListObjectsCompleted(err) - if err != nil { - return result, err + for pager.NextPage(ctx) { + resp := pager.PageResponse() + result = len(resp.ContainerListBlobFlatSegmentResult.Segment.BlobItems) > 0 + if result { + break + } } - result = len(listBlob.Segment.BlobItems) > 0 + + err := pager.Err() + metric.AZListObjectsCompleted(err) return result, err } -func (fs *AzureBlobFs) handleMultipartUpload(ctx context.Context, reader io.Reader, blockBlobURL *azblob.BlockBlobURL, - httpHeaders *azblob.BlobHTTPHeaders) error { +func (fs *AzureBlobFs) handleMultipartUpload(ctx context.Context, reader io.Reader, + blockBlob azblob.BlockBlobClient, httpHeaders *azblob.BlobHTTPHeaders, +) error { partSize := fs.config.UploadPartSize guard := make(chan struct{}, fs.config.UploadConcurrency) blockCtxTimeout := time.Duration(fs.config.UploadPartSize/(1024*1024)) * time.Minute @@ -946,12 +926,14 @@ func (fs *AzureBlobFs) handleMultipartUpload(ctx context.Context, reader io.Read wg.Add(1) go func(blockID string, buf []byte, bufSize int) { defer wg.Done() - bufferReader := bytes.NewReader(buf[:bufSize]) + + bufferReader := &bytesReaderWrapper{ + Reader: bytes.NewReader(buf[:bufSize]), + } innerCtx, cancelFn := context.WithDeadline(poolCtx, time.Now().Add(blockCtxTimeout)) defer cancelFn() - _, err := blockBlobURL.StageBlock(innerCtx, blockID, bufferReader, azblob.LeaseAccessConditions{}, nil, - azblob.ClientProvidedKeyOptions{}) + _, err := blockBlob.StageBlock(innerCtx, blockID, bufferReader, &azblob.StageBlockOptions{}) if err != nil { errOnce.Do(func() { poolError = err @@ -972,8 +954,14 @@ func (fs *AzureBlobFs) handleMultipartUpload(ctx context.Context, reader io.Read return poolError } - _, err := blockBlobURL.CommitBlockList(ctx, blocks, *httpHeaders, azblob.Metadata{}, azblob.BlobAccessConditions{}, - azblob.AccessTierType(fs.config.AccessTier), nil, azblob.ClientProvidedKeyOptions{}) + commitOptions := azblob.CommitBlockListOptions{ + BlobHTTPHeaders: httpHeaders, + } + if fs.config.AccessTier != "" { + commitOptions.Tier = (*azblob.AccessTier)(&fs.config.AccessTier) + } + + _, err := blockBlob.CommitBlockList(ctx, blocks, &commitOptions) return err } @@ -999,6 +987,14 @@ func (fs *AzureBlobFs) incrementBlockID(blockID []byte) { } } +func (fs *AzureBlobFs) getCopyOptions() *azblob.StartCopyBlobOptions { + copyOptions := &azblob.StartCopyBlobOptions{} + if fs.config.AccessTier != "" { + copyOptions.Tier = (*azblob.AccessTier)(&fs.config.AccessTier) + } + return copyOptions +} + func (fs *AzureBlobFs) getStorageID() string { if fs.config.Endpoint != "" { if !strings.HasSuffix(fs.config.Endpoint, "/") { @@ -1009,6 +1005,14 @@ func (fs *AzureBlobFs) getStorageID() string { return fmt.Sprintf("azblob://%v", fs.config.Container) } +type bytesReaderWrapper struct { + *bytes.Reader +} + +func (b *bytesReaderWrapper) Close() error { + return nil +} + type bufferAllocator struct { sync.Mutex available [][]byte