From d31cccf85f711e3efac9d8315746c515b4577d14 Mon Sep 17 00:00:00 2001
From: Nicola Murino <nicola.murino@gmail.com>
Date: Sun, 20 Feb 2022 14:43:24 +0100
Subject: [PATCH] azblob: switch to the new azure-go-sdk
Signed-off-by: Nicola Murino <nicola.murino@gmail.com>
---
go.mod | 6 +-
go.sum | 24 +--
util/util.go | 26 ++-
vfs/azblobfs.go | 554 ++++++++++++++++++++++++------------------------
4 files changed, 319 insertions(+), 291 deletions(-)
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