azblobfs.go 36 KB

123456789101112131415161718192021222324252627282930313233343536373839404142434445464748495051525354555657585960616263646566676869707172737475767778798081828384858687888990919293949596979899100101102103104105106107108109110111112113114115116117118119120121122123124125126127128129130131132133134135136137138139140141142143144145146147148149150151152153154155156157158159160161162163164165166167168169170171172173174175176177178179180181182183184185186187188189190191192193194195196197198199200201202203204205206207208209210211212213214215216217218219220221222223224225226227228229230231232233234235236237238239240241242243244245246247248249250251252253254255256257258259260261262263264265266267268269270271272273274275276277278279280281282283284285286287288289290291292293294295296297298299300301302303304305306307308309310311312313314315316317318319320321322323324325326327328329330331332333334335336337338339340341342343344345346347348349350351352353354355356357358359360361362363364365366367368369370371372373374375376377378379380381382383384385386387388389390391392393394395396397398399400401402403404405406407408409410411412413414415416417418419420421422423424425426427428429430431432433434435436437438439440441442443444445446447448449450451452453454455456457458459460461462463464465466467468469470471472473474475476477478479480481482483484485486487488489490491492493494495496497498499500501502503504505506507508509510511512513514515516517518519520521522523524525526527528529530531532533534535536537538539540541542543544545546547548549550551552553554555556557558559560561562563564565566567568569570571572573574575576577578579580581582583584585586587588589590591592593594595596597598599600601602603604605606607608609610611612613614615616617618619620621622623624625626627628629630631632633634635636637638639640641642643644645646647648649650651652653654655656657658659660661662663664665666667668669670671672673674675676677678679680681682683684685686687688689690691692693694695696697698699700701702703704705706707708709710711712713714715716717718719720721722723724725726727728729730731732733734735736737738739740741742743744745746747748749750751752753754755756757758759760761762763764765766767768769770771772773774775776777778779780781782783784785786787788789790791792793794795796797798799800801802803804805806807808809810811812813814815816817818819820821822823824825826827828829830831832833834835836837838839840841842843844845846847848849850851852853854855856857858859860861862863864865866867868869870871872873874875876877878879880881882883884885886887888889890891892893894895896897898899900901902903904905906907908909910911912913914915916917918919920921922923924925926927928929930931932933934935936937938939940941942943944945946947948949950951952953954955956957958959960961962963964965966967968969970971972973974975976977978979980981982983984985986987988989990991992993994995996997998999100010011002100310041005100610071008100910101011101210131014101510161017101810191020102110221023102410251026102710281029103010311032103310341035103610371038103910401041104210431044104510461047104810491050105110521053105410551056105710581059106010611062106310641065106610671068106910701071107210731074107510761077107810791080108110821083108410851086108710881089109010911092109310941095109610971098109911001101110211031104110511061107110811091110111111121113111411151116111711181119112011211122112311241125112611271128112911301131113211331134113511361137113811391140114111421143114411451146114711481149115011511152115311541155115611571158115911601161116211631164116511661167116811691170117111721173117411751176117711781179118011811182118311841185118611871188118911901191119211931194119511961197119811991200120112021203120412051206120712081209121012111212121312141215121612171218121912201221122212231224122512261227122812291230123112321233123412351236123712381239124012411242124312441245124612471248124912501251125212531254125512561257125812591260126112621263126412651266126712681269127012711272127312741275127612771278127912801281128212831284
  1. // Copyright (C) 2019 Nicola Murino
  2. //
  3. // This program is free software: you can redistribute it and/or modify
  4. // it under the terms of the GNU Affero General Public License as published
  5. // by the Free Software Foundation, version 3.
  6. //
  7. // This program is distributed in the hope that it will be useful,
  8. // but WITHOUT ANY WARRANTY; without even the implied warranty of
  9. // MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE. See the
  10. // GNU Affero General Public License for more details.
  11. //
  12. // You should have received a copy of the GNU Affero General Public License
  13. // along with this program. If not, see <https://www.gnu.org/licenses/>.
  14. //go:build !noazblob
  15. // +build !noazblob
  16. package vfs
  17. import (
  18. "bytes"
  19. "context"
  20. "encoding/base64"
  21. "errors"
  22. "fmt"
  23. "io"
  24. "mime"
  25. "net/http"
  26. "os"
  27. "path"
  28. "path/filepath"
  29. "strconv"
  30. "strings"
  31. "sync"
  32. "sync/atomic"
  33. "time"
  34. "github.com/Azure/azure-sdk-for-go/sdk/azcore"
  35. "github.com/Azure/azure-sdk-for-go/sdk/azcore/policy"
  36. "github.com/Azure/azure-sdk-for-go/sdk/azcore/runtime"
  37. "github.com/Azure/azure-sdk-for-go/sdk/azcore/to"
  38. "github.com/Azure/azure-sdk-for-go/sdk/storage/azblob/blob"
  39. "github.com/Azure/azure-sdk-for-go/sdk/storage/azblob/blockblob"
  40. "github.com/Azure/azure-sdk-for-go/sdk/storage/azblob/container"
  41. "github.com/eikenb/pipeat"
  42. "github.com/google/uuid"
  43. "github.com/pkg/sftp"
  44. "github.com/drakkan/sftpgo/v2/internal/logger"
  45. "github.com/drakkan/sftpgo/v2/internal/metric"
  46. "github.com/drakkan/sftpgo/v2/internal/util"
  47. "github.com/drakkan/sftpgo/v2/internal/version"
  48. )
  49. const (
  50. azureDefaultEndpoint = "blob.core.windows.net"
  51. azFolderKey = "hdi_isfolder"
  52. )
  53. var (
  54. azureBlobDefaultPageSize = int32(5000)
  55. )
  56. // AzureBlobFs is a Fs implementation for Azure Blob storage.
  57. type AzureBlobFs struct {
  58. connectionID string
  59. localTempDir string
  60. // if not empty this fs is mouted as virtual folder in the specified path
  61. mountPath string
  62. config *AzBlobFsConfig
  63. containerClient *container.Client
  64. ctxTimeout time.Duration
  65. ctxLongTimeout time.Duration
  66. }
  67. func init() {
  68. version.AddFeature("+azblob")
  69. }
  70. // NewAzBlobFs returns an AzBlobFs object that allows to interact with Azure Blob storage
  71. func NewAzBlobFs(connectionID, localTempDir, mountPath string, config AzBlobFsConfig) (Fs, error) {
  72. if localTempDir == "" {
  73. localTempDir = getLocalTempDir()
  74. }
  75. fs := &AzureBlobFs{
  76. connectionID: connectionID,
  77. localTempDir: localTempDir,
  78. mountPath: getMountPath(mountPath),
  79. config: &config,
  80. ctxTimeout: 30 * time.Second,
  81. ctxLongTimeout: 90 * time.Second,
  82. }
  83. if err := fs.config.validate(); err != nil {
  84. return fs, err
  85. }
  86. if err := fs.config.tryDecrypt(); err != nil {
  87. return fs, err
  88. }
  89. fs.setConfigDefaults()
  90. if fs.config.SASURL.GetPayload() != "" {
  91. return fs.initFromSASURL()
  92. }
  93. credential, err := blob.NewSharedKeyCredential(fs.config.AccountName, fs.config.AccountKey.GetPayload())
  94. if err != nil {
  95. return fs, fmt.Errorf("invalid credentials: %v", err)
  96. }
  97. var endpoint string
  98. if fs.config.UseEmulator {
  99. endpoint = fmt.Sprintf("%s/%s", fs.config.Endpoint, fs.config.AccountName)
  100. } else {
  101. endpoint = fmt.Sprintf("https://%s.%s/", fs.config.AccountName, fs.config.Endpoint)
  102. }
  103. containerURL := runtime.JoinPaths(endpoint, fs.config.Container)
  104. svc, err := container.NewClientWithSharedKeyCredential(containerURL, credential, getAzContainerClientOptions())
  105. if err != nil {
  106. return fs, fmt.Errorf("invalid credentials: %v", err)
  107. }
  108. fs.containerClient = svc
  109. return fs, err
  110. }
  111. func (fs *AzureBlobFs) initFromSASURL() (Fs, error) {
  112. parts, err := blob.ParseURL(fs.config.SASURL.GetPayload())
  113. if err != nil {
  114. return fs, fmt.Errorf("invalid SAS URL: %w", err)
  115. }
  116. if parts.BlobName != "" {
  117. return fs, fmt.Errorf("SAS URL with blob name not supported")
  118. }
  119. if parts.ContainerName != "" {
  120. if fs.config.Container != "" && fs.config.Container != parts.ContainerName {
  121. return fs, fmt.Errorf("container name in SAS URL %q and container provided %q do not match",
  122. parts.ContainerName, fs.config.Container)
  123. }
  124. svc, err := container.NewClientWithNoCredential(fs.config.SASURL.GetPayload(), getAzContainerClientOptions())
  125. if err != nil {
  126. return fs, fmt.Errorf("invalid credentials: %v", err)
  127. }
  128. fs.config.Container = parts.ContainerName
  129. fs.containerClient = svc
  130. return fs, nil
  131. }
  132. if fs.config.Container == "" {
  133. return fs, errors.New("container is required with this SAS URL")
  134. }
  135. sasURL := runtime.JoinPaths(fs.config.SASURL.GetPayload(), fs.config.Container)
  136. svc, err := container.NewClientWithNoCredential(sasURL, getAzContainerClientOptions())
  137. if err != nil {
  138. return fs, fmt.Errorf("invalid credentials: %v", err)
  139. }
  140. fs.containerClient = svc
  141. return fs, nil
  142. }
  143. // Name returns the name for the Fs implementation
  144. func (fs *AzureBlobFs) Name() string {
  145. if !fs.config.SASURL.IsEmpty() {
  146. return fmt.Sprintf("%s with SAS URL, container %q", azBlobFsName, fs.config.Container)
  147. }
  148. return fmt.Sprintf("%s container %q", azBlobFsName, fs.config.Container)
  149. }
  150. // ConnectionID returns the connection ID associated to this Fs implementation
  151. func (fs *AzureBlobFs) ConnectionID() string {
  152. return fs.connectionID
  153. }
  154. // Stat returns a FileInfo describing the named file
  155. func (fs *AzureBlobFs) Stat(name string) (os.FileInfo, error) {
  156. if name == "" || name == "/" || name == "." {
  157. return NewFileInfo(name, true, 0, time.Unix(0, 0), false), nil
  158. }
  159. if fs.config.KeyPrefix == name+"/" {
  160. return NewFileInfo(name, true, 0, time.Unix(0, 0), false), nil
  161. }
  162. attrs, err := fs.headObject(name)
  163. if err == nil {
  164. contentType := util.GetStringFromPointer(attrs.ContentType)
  165. isDir := checkDirectoryMarkers(contentType, attrs.Metadata)
  166. lastModified := util.GetTimeFromPointer(attrs.LastModified)
  167. if val := getAzureLastModified(attrs.Metadata); val > 0 {
  168. lastModified = util.GetTimeFromMsecSinceEpoch(val)
  169. }
  170. return NewFileInfo(name, isDir, util.GetIntFromPointer(attrs.ContentLength), lastModified, false), nil
  171. }
  172. if !fs.IsNotExist(err) {
  173. return nil, err
  174. }
  175. // now check if this is a prefix (virtual directory)
  176. hasContents, err := fs.hasContents(name)
  177. if err != nil {
  178. return nil, err
  179. }
  180. if hasContents {
  181. return NewFileInfo(name, true, 0, time.Unix(0, 0), false), nil
  182. }
  183. return nil, os.ErrNotExist
  184. }
  185. // Lstat returns a FileInfo describing the named file
  186. func (fs *AzureBlobFs) Lstat(name string) (os.FileInfo, error) {
  187. return fs.Stat(name)
  188. }
  189. // Open opens the named file for reading
  190. func (fs *AzureBlobFs) Open(name string, offset int64) (File, PipeReader, func(), error) {
  191. r, w, err := pipeat.PipeInDir(fs.localTempDir)
  192. if err != nil {
  193. return nil, nil, nil, err
  194. }
  195. p := NewPipeReader(r)
  196. ctx, cancelFn := context.WithCancel(context.Background())
  197. go func() {
  198. defer cancelFn()
  199. blockBlob := fs.containerClient.NewBlockBlobClient(name)
  200. err := fs.handleMultipartDownload(ctx, blockBlob, offset, w, p)
  201. w.CloseWithError(err) //nolint:errcheck
  202. fsLog(fs, logger.LevelDebug, "download completed, path: %q size: %v, err: %+v", name, w.GetWrittenBytes(), err)
  203. metric.AZTransferCompleted(w.GetWrittenBytes(), 1, err)
  204. }()
  205. return nil, p, cancelFn, nil
  206. }
  207. // Create creates or opens the named file for writing
  208. func (fs *AzureBlobFs) Create(name string, flag, checks int) (File, PipeWriter, func(), error) {
  209. if checks&CheckParentDir != 0 {
  210. _, err := fs.Stat(path.Dir(name))
  211. if err != nil {
  212. return nil, nil, nil, err
  213. }
  214. }
  215. r, w, err := pipeat.PipeInDir(fs.localTempDir)
  216. if err != nil {
  217. return nil, nil, nil, err
  218. }
  219. ctx, cancelFn := context.WithCancel(context.Background())
  220. var p PipeWriter
  221. if checks&CheckResume != 0 {
  222. p = newPipeWriterAtOffset(w, 0)
  223. } else {
  224. p = NewPipeWriter(w)
  225. }
  226. headers := blob.HTTPHeaders{}
  227. var contentType string
  228. var metadata map[string]*string
  229. if flag == -1 {
  230. contentType = dirMimeType
  231. metadata = map[string]*string{
  232. azFolderKey: util.NilIfEmpty("true"),
  233. }
  234. } else {
  235. contentType = mime.TypeByExtension(path.Ext(name))
  236. }
  237. if contentType != "" {
  238. headers.BlobContentType = &contentType
  239. }
  240. go func() {
  241. defer cancelFn()
  242. blockBlob := fs.containerClient.NewBlockBlobClient(name)
  243. err := fs.handleMultipartUpload(ctx, r, blockBlob, &headers, metadata)
  244. r.CloseWithError(err) //nolint:errcheck
  245. p.Done(err)
  246. fsLog(fs, logger.LevelDebug, "upload completed, path: %q, readed bytes: %v, err: %+v", name, r.GetReadedBytes(), err)
  247. metric.AZTransferCompleted(r.GetReadedBytes(), 0, err)
  248. }()
  249. if checks&CheckResume != 0 {
  250. readCh := make(chan error, 1)
  251. go func() {
  252. n, err := fs.downloadToWriter(name, p)
  253. pw := p.(*pipeWriterAtOffset)
  254. pw.offset = 0
  255. pw.writeOffset = n
  256. readCh <- err
  257. }()
  258. err = <-readCh
  259. if err != nil {
  260. cancelFn()
  261. p.Close()
  262. fsLog(fs, logger.LevelDebug, "download before resume failed, writer closed and read cancelled")
  263. return nil, nil, nil, err
  264. }
  265. }
  266. if uploadMode&16 != 0 {
  267. return nil, p, nil, nil
  268. }
  269. return nil, p, cancelFn, nil
  270. }
  271. // Rename renames (moves) source to target.
  272. func (fs *AzureBlobFs) Rename(source, target string) (int, int64, error) {
  273. if source == target {
  274. return -1, -1, nil
  275. }
  276. _, err := fs.Stat(path.Dir(target))
  277. if err != nil {
  278. return -1, -1, err
  279. }
  280. fi, err := fs.Stat(source)
  281. if err != nil {
  282. return -1, -1, err
  283. }
  284. return fs.renameInternal(source, target, fi, 0)
  285. }
  286. // Remove removes the named file or (empty) directory.
  287. func (fs *AzureBlobFs) Remove(name string, isDir bool) error {
  288. if isDir {
  289. hasContents, err := fs.hasContents(name)
  290. if err != nil {
  291. return err
  292. }
  293. if hasContents {
  294. return fmt.Errorf("cannot remove non empty directory: %q", name)
  295. }
  296. }
  297. ctx, cancelFn := context.WithDeadline(context.Background(), time.Now().Add(fs.ctxTimeout))
  298. defer cancelFn()
  299. blobBlock := fs.containerClient.NewBlockBlobClient(name)
  300. var deletSnapshots blob.DeleteSnapshotsOptionType
  301. if !isDir {
  302. deletSnapshots = blob.DeleteSnapshotsOptionTypeInclude
  303. }
  304. _, err := blobBlock.Delete(ctx, &blob.DeleteOptions{
  305. DeleteSnapshots: &deletSnapshots,
  306. })
  307. if err != nil && isDir {
  308. if fs.isBadRequestError(err) {
  309. deletSnapshots = blob.DeleteSnapshotsOptionTypeInclude
  310. _, err = blobBlock.Delete(ctx, &blob.DeleteOptions{
  311. DeleteSnapshots: &deletSnapshots,
  312. })
  313. }
  314. }
  315. metric.AZDeleteObjectCompleted(err)
  316. return err
  317. }
  318. // Mkdir creates a new directory with the specified name and default permissions
  319. func (fs *AzureBlobFs) Mkdir(name string) error {
  320. _, err := fs.Stat(name)
  321. if !fs.IsNotExist(err) {
  322. return err
  323. }
  324. return fs.mkdirInternal(name)
  325. }
  326. // Symlink creates source as a symbolic link to target.
  327. func (*AzureBlobFs) Symlink(_, _ string) error {
  328. return ErrVfsUnsupported
  329. }
  330. // Readlink returns the destination of the named symbolic link
  331. func (*AzureBlobFs) Readlink(_ string) (string, error) {
  332. return "", ErrVfsUnsupported
  333. }
  334. // Chown changes the numeric uid and gid of the named file.
  335. func (*AzureBlobFs) Chown(_ string, _ int, _ int) error {
  336. return ErrVfsUnsupported
  337. }
  338. // Chmod changes the mode of the named file to mode.
  339. func (*AzureBlobFs) Chmod(_ string, _ os.FileMode) error {
  340. return ErrVfsUnsupported
  341. }
  342. // Chtimes changes the access and modification times of the named file.
  343. func (fs *AzureBlobFs) Chtimes(name string, _, mtime time.Time, isUploading bool) error {
  344. if isUploading {
  345. return nil
  346. }
  347. props, err := fs.headObject(name)
  348. if err != nil {
  349. return err
  350. }
  351. metadata := props.Metadata
  352. if metadata == nil {
  353. metadata = make(map[string]*string)
  354. }
  355. metadata[lastModifiedField] = to.Ptr(strconv.FormatInt(mtime.UnixMilli(), 10))
  356. ctx, cancelFn := context.WithDeadline(context.Background(), time.Now().Add(fs.ctxTimeout))
  357. defer cancelFn()
  358. _, err = fs.containerClient.NewBlockBlobClient(name).SetMetadata(ctx, metadata, &blob.SetMetadataOptions{})
  359. return err
  360. }
  361. // Truncate changes the size of the named file.
  362. // Truncate by path is not supported, while truncating an opened
  363. // file is handled inside base transfer
  364. func (*AzureBlobFs) Truncate(_ string, _ int64) error {
  365. return ErrVfsUnsupported
  366. }
  367. // ReadDir reads the directory named by dirname and returns
  368. // a list of directory entries.
  369. func (fs *AzureBlobFs) ReadDir(dirname string) (DirLister, error) {
  370. // dirname must be already cleaned
  371. prefix := fs.getPrefix(dirname)
  372. pager := fs.containerClient.NewListBlobsHierarchyPager("/", &container.ListBlobsHierarchyOptions{
  373. Include: container.ListBlobsInclude{
  374. Metadata: true,
  375. },
  376. Prefix: &prefix,
  377. MaxResults: &azureBlobDefaultPageSize,
  378. })
  379. return &azureBlobDirLister{
  380. paginator: pager,
  381. timeout: fs.ctxTimeout,
  382. prefix: prefix,
  383. prefixes: make(map[string]bool),
  384. }, nil
  385. }
  386. // IsUploadResumeSupported returns true if resuming uploads is supported.
  387. // Resuming uploads is not supported on Azure Blob
  388. func (*AzureBlobFs) IsUploadResumeSupported() bool {
  389. return false
  390. }
  391. // IsConditionalUploadResumeSupported returns if resuming uploads is supported
  392. // for the specified size
  393. func (*AzureBlobFs) IsConditionalUploadResumeSupported(size int64) bool {
  394. return size <= resumeMaxSize
  395. }
  396. // IsAtomicUploadSupported returns true if atomic upload is supported.
  397. // Azure Blob uploads are already atomic, we don't need to upload to a temporary
  398. // file
  399. func (*AzureBlobFs) IsAtomicUploadSupported() bool {
  400. return false
  401. }
  402. // IsNotExist returns a boolean indicating whether the error is known to
  403. // report that a file or directory does not exist
  404. func (*AzureBlobFs) IsNotExist(err error) bool {
  405. if err == nil {
  406. return false
  407. }
  408. var respErr *azcore.ResponseError
  409. if errors.As(err, &respErr) {
  410. return respErr.StatusCode == http.StatusNotFound
  411. }
  412. // os.ErrNotExist can be returned internally by fs.Stat
  413. return errors.Is(err, os.ErrNotExist)
  414. }
  415. // IsPermission returns a boolean indicating whether the error is known to
  416. // report that permission is denied.
  417. func (*AzureBlobFs) IsPermission(err error) bool {
  418. if err == nil {
  419. return false
  420. }
  421. var respErr *azcore.ResponseError
  422. if errors.As(err, &respErr) {
  423. return respErr.StatusCode == http.StatusForbidden || respErr.StatusCode == http.StatusUnauthorized
  424. }
  425. return false
  426. }
  427. // IsNotSupported returns true if the error indicate an unsupported operation
  428. func (*AzureBlobFs) IsNotSupported(err error) bool {
  429. if err == nil {
  430. return false
  431. }
  432. return errors.Is(err, ErrVfsUnsupported)
  433. }
  434. func (*AzureBlobFs) isBadRequestError(err error) bool {
  435. if err == nil {
  436. return false
  437. }
  438. var respErr *azcore.ResponseError
  439. if errors.As(err, &respErr) {
  440. return respErr.StatusCode == http.StatusBadRequest
  441. }
  442. return false
  443. }
  444. // CheckRootPath creates the specified local root directory if it does not exists
  445. func (fs *AzureBlobFs) CheckRootPath(username string, uid int, gid int) bool {
  446. // we need a local directory for temporary files
  447. osFs := NewOsFs(fs.ConnectionID(), fs.localTempDir, "", nil)
  448. return osFs.CheckRootPath(username, uid, gid)
  449. }
  450. // ScanRootDirContents returns the number of files contained in the bucket,
  451. // and their size
  452. func (fs *AzureBlobFs) ScanRootDirContents() (int, int64, error) {
  453. return fs.GetDirSize(fs.config.KeyPrefix)
  454. }
  455. // GetDirSize returns the number of files and the size for a folder
  456. // including any subfolders
  457. func (fs *AzureBlobFs) GetDirSize(dirname string) (int, int64, error) {
  458. numFiles := 0
  459. size := int64(0)
  460. prefix := fs.getPrefix(dirname)
  461. pager := fs.containerClient.NewListBlobsFlatPager(&container.ListBlobsFlatOptions{
  462. Include: container.ListBlobsInclude{
  463. Metadata: true,
  464. },
  465. Prefix: &prefix,
  466. MaxResults: &azureBlobDefaultPageSize,
  467. })
  468. for pager.More() {
  469. ctx, cancelFn := context.WithDeadline(context.Background(), time.Now().Add(fs.ctxTimeout))
  470. defer cancelFn()
  471. resp, err := pager.NextPage(ctx)
  472. if err != nil {
  473. metric.AZListObjectsCompleted(err)
  474. return numFiles, size, err
  475. }
  476. for _, blobItem := range resp.ListBlobsFlatSegmentResponse.Segment.BlobItems {
  477. if blobItem.Properties != nil {
  478. contentType := util.GetStringFromPointer(blobItem.Properties.ContentType)
  479. isDir := checkDirectoryMarkers(contentType, blobItem.Metadata)
  480. blobSize := util.GetIntFromPointer(blobItem.Properties.ContentLength)
  481. if isDir && blobSize == 0 {
  482. continue
  483. }
  484. numFiles++
  485. size += blobSize
  486. }
  487. }
  488. fsLog(fs, logger.LevelDebug, "scan in progress for %q, files: %d, size: %d", dirname, numFiles, size)
  489. }
  490. metric.AZListObjectsCompleted(nil)
  491. return numFiles, size, nil
  492. }
  493. // GetAtomicUploadPath returns the path to use for an atomic upload.
  494. // Azure Blob Storage uploads are already atomic, we never call this method
  495. func (*AzureBlobFs) GetAtomicUploadPath(_ string) string {
  496. return ""
  497. }
  498. // GetRelativePath returns the path for a file relative to the user's home dir.
  499. // This is the path as seen by SFTPGo users
  500. func (fs *AzureBlobFs) GetRelativePath(name string) string {
  501. rel := path.Clean(name)
  502. if rel == "." {
  503. rel = ""
  504. }
  505. if !path.IsAbs(rel) {
  506. rel = "/" + rel
  507. }
  508. if fs.config.KeyPrefix != "" {
  509. if !strings.HasPrefix(rel, "/"+fs.config.KeyPrefix) {
  510. rel = "/"
  511. }
  512. rel = path.Clean("/" + strings.TrimPrefix(rel, "/"+fs.config.KeyPrefix))
  513. }
  514. if fs.mountPath != "" {
  515. rel = path.Join(fs.mountPath, rel)
  516. }
  517. return rel
  518. }
  519. // Walk walks the file tree rooted at root, calling walkFn for each file or
  520. // directory in the tree, including root
  521. func (fs *AzureBlobFs) Walk(root string, walkFn filepath.WalkFunc) error {
  522. prefix := fs.getPrefix(root)
  523. pager := fs.containerClient.NewListBlobsFlatPager(&container.ListBlobsFlatOptions{
  524. Include: container.ListBlobsInclude{
  525. Metadata: true,
  526. },
  527. Prefix: &prefix,
  528. MaxResults: &azureBlobDefaultPageSize,
  529. })
  530. for pager.More() {
  531. ctx, cancelFn := context.WithDeadline(context.Background(), time.Now().Add(fs.ctxTimeout))
  532. defer cancelFn()
  533. resp, err := pager.NextPage(ctx)
  534. if err != nil {
  535. metric.AZListObjectsCompleted(err)
  536. return err
  537. }
  538. for _, blobItem := range resp.ListBlobsFlatSegmentResponse.Segment.BlobItems {
  539. name := util.GetStringFromPointer(blobItem.Name)
  540. if fs.isEqual(name, prefix) {
  541. continue
  542. }
  543. blobSize := int64(0)
  544. lastModified := time.Unix(0, 0)
  545. isDir := false
  546. if blobItem.Properties != nil {
  547. contentType := util.GetStringFromPointer(blobItem.Properties.ContentType)
  548. isDir = checkDirectoryMarkers(contentType, blobItem.Metadata)
  549. blobSize = util.GetIntFromPointer(blobItem.Properties.ContentLength)
  550. lastModified = util.GetTimeFromPointer(blobItem.Properties.LastModified)
  551. if val := getAzureLastModified(blobItem.Metadata); val > 0 {
  552. lastModified = util.GetTimeFromMsecSinceEpoch(val)
  553. }
  554. }
  555. err := walkFn(name, NewFileInfo(name, isDir, blobSize, lastModified, false), nil)
  556. if err != nil {
  557. return err
  558. }
  559. }
  560. }
  561. metric.AZListObjectsCompleted(nil)
  562. return walkFn(root, NewFileInfo(root, true, 0, time.Unix(0, 0), false), nil)
  563. }
  564. // Join joins any number of path elements into a single path
  565. func (*AzureBlobFs) Join(elem ...string) string {
  566. return strings.TrimPrefix(path.Join(elem...), "/")
  567. }
  568. // HasVirtualFolders returns true if folders are emulated
  569. func (*AzureBlobFs) HasVirtualFolders() bool {
  570. return true
  571. }
  572. // ResolvePath returns the matching filesystem path for the specified sftp path
  573. func (fs *AzureBlobFs) ResolvePath(virtualPath string) (string, error) {
  574. if fs.mountPath != "" {
  575. virtualPath = strings.TrimPrefix(virtualPath, fs.mountPath)
  576. }
  577. if !path.IsAbs(virtualPath) {
  578. virtualPath = path.Clean("/" + virtualPath)
  579. }
  580. return fs.Join(fs.config.KeyPrefix, strings.TrimPrefix(virtualPath, "/")), nil
  581. }
  582. // CopyFile implements the FsFileCopier interface
  583. func (fs *AzureBlobFs) CopyFile(source, target string, srcSize int64) (int, int64, error) {
  584. numFiles := 1
  585. sizeDiff := srcSize
  586. attrs, err := fs.headObject(target)
  587. if err == nil {
  588. sizeDiff -= util.GetIntFromPointer(attrs.ContentLength)
  589. numFiles = 0
  590. } else {
  591. if !fs.IsNotExist(err) {
  592. return 0, 0, err
  593. }
  594. }
  595. if err := fs.copyFileInternal(source, target); err != nil {
  596. return 0, 0, err
  597. }
  598. return numFiles, sizeDiff, nil
  599. }
  600. func (fs *AzureBlobFs) headObject(name string) (blob.GetPropertiesResponse, error) {
  601. ctx, cancelFn := context.WithDeadline(context.Background(), time.Now().Add(fs.ctxTimeout))
  602. defer cancelFn()
  603. resp, err := fs.containerClient.NewBlockBlobClient(name).GetProperties(ctx, &blob.GetPropertiesOptions{})
  604. metric.AZHeadObjectCompleted(err)
  605. return resp, err
  606. }
  607. // GetMimeType returns the content type
  608. func (fs *AzureBlobFs) GetMimeType(name string) (string, error) {
  609. response, err := fs.headObject(name)
  610. if err != nil {
  611. return "", err
  612. }
  613. return util.GetStringFromPointer(response.ContentType), nil
  614. }
  615. // Close closes the fs
  616. func (*AzureBlobFs) Close() error {
  617. return nil
  618. }
  619. // GetAvailableDiskSize returns the available size for the specified path
  620. func (*AzureBlobFs) GetAvailableDiskSize(_ string) (*sftp.StatVFS, error) {
  621. return nil, ErrStorageSizeUnavailable
  622. }
  623. func (*AzureBlobFs) getPrefix(name string) string {
  624. prefix := ""
  625. if name != "" && name != "." {
  626. prefix = strings.TrimPrefix(name, "/")
  627. if !strings.HasSuffix(prefix, "/") {
  628. prefix += "/"
  629. }
  630. }
  631. return prefix
  632. }
  633. func (fs *AzureBlobFs) isEqual(key string, virtualName string) bool {
  634. if key == virtualName {
  635. return true
  636. }
  637. if key == virtualName+"/" {
  638. return true
  639. }
  640. if key+"/" == virtualName {
  641. return true
  642. }
  643. return false
  644. }
  645. func (fs *AzureBlobFs) setConfigDefaults() {
  646. if fs.config.Endpoint == "" {
  647. fs.config.Endpoint = azureDefaultEndpoint
  648. }
  649. if fs.config.UploadPartSize == 0 {
  650. fs.config.UploadPartSize = 5
  651. }
  652. if fs.config.UploadPartSize < 1024*1024 {
  653. fs.config.UploadPartSize *= 1024 * 1024
  654. }
  655. if fs.config.UploadConcurrency == 0 {
  656. fs.config.UploadConcurrency = 5
  657. }
  658. if fs.config.DownloadPartSize == 0 {
  659. fs.config.DownloadPartSize = 5
  660. }
  661. if fs.config.DownloadPartSize < 1024*1024 {
  662. fs.config.DownloadPartSize *= 1024 * 1024
  663. }
  664. if fs.config.DownloadConcurrency == 0 {
  665. fs.config.DownloadConcurrency = 5
  666. }
  667. }
  668. func (fs *AzureBlobFs) copyFileInternal(source, target string) error {
  669. ctx, cancelFn := context.WithDeadline(context.Background(), time.Now().Add(fs.ctxLongTimeout))
  670. defer cancelFn()
  671. srcBlob := fs.containerClient.NewBlockBlobClient(source)
  672. dstBlob := fs.containerClient.NewBlockBlobClient(target)
  673. resp, err := dstBlob.StartCopyFromURL(ctx, srcBlob.URL(), fs.getCopyOptions())
  674. if err != nil {
  675. metric.AZCopyObjectCompleted(err)
  676. return err
  677. }
  678. copyStatus := blob.CopyStatusType(util.GetStringFromPointer((*string)(resp.CopyStatus)))
  679. nErrors := 0
  680. for copyStatus == blob.CopyStatusTypePending {
  681. // Poll until the copy is complete.
  682. time.Sleep(500 * time.Millisecond)
  683. resp, err := dstBlob.GetProperties(ctx, &blob.GetPropertiesOptions{})
  684. if err != nil {
  685. // A GetProperties failure may be transient, so allow a couple
  686. // of them before giving up.
  687. nErrors++
  688. if ctx.Err() != nil || nErrors == 3 {
  689. metric.AZCopyObjectCompleted(err)
  690. return err
  691. }
  692. } else {
  693. copyStatus = blob.CopyStatusType(util.GetStringFromPointer((*string)(resp.CopyStatus)))
  694. }
  695. }
  696. if copyStatus != blob.CopyStatusTypeSuccess {
  697. err := fmt.Errorf("copy failed with status: %s", copyStatus)
  698. metric.AZCopyObjectCompleted(err)
  699. return err
  700. }
  701. metric.AZCopyObjectCompleted(nil)
  702. return nil
  703. }
  704. func (fs *AzureBlobFs) renameInternal(source, target string, fi os.FileInfo, recursion int) (int, int64, error) {
  705. var numFiles int
  706. var filesSize int64
  707. if fi.IsDir() {
  708. if renameMode == 0 {
  709. hasContents, err := fs.hasContents(source)
  710. if err != nil {
  711. return numFiles, filesSize, err
  712. }
  713. if hasContents {
  714. return numFiles, filesSize, fmt.Errorf("%w: cannot rename non empty directory: %q", ErrVfsUnsupported, source)
  715. }
  716. }
  717. if err := fs.mkdirInternal(target); err != nil {
  718. return numFiles, filesSize, err
  719. }
  720. if renameMode == 1 {
  721. files, size, err := doRecursiveRename(fs, source, target, fs.renameInternal, recursion)
  722. numFiles += files
  723. filesSize += size
  724. if err != nil {
  725. return numFiles, filesSize, err
  726. }
  727. }
  728. } else {
  729. if err := fs.copyFileInternal(source, target); err != nil {
  730. return numFiles, filesSize, err
  731. }
  732. numFiles++
  733. filesSize += fi.Size()
  734. }
  735. err := fs.skipNotExistErr(fs.Remove(source, fi.IsDir()))
  736. return numFiles, filesSize, err
  737. }
  738. func (fs *AzureBlobFs) skipNotExistErr(err error) error {
  739. if fs.IsNotExist(err) {
  740. return nil
  741. }
  742. return err
  743. }
  744. func (fs *AzureBlobFs) mkdirInternal(name string) error {
  745. _, w, _, err := fs.Create(name, -1, 0)
  746. if err != nil {
  747. return err
  748. }
  749. return w.Close()
  750. }
  751. func (fs *AzureBlobFs) hasContents(name string) (bool, error) {
  752. result := false
  753. prefix := fs.getPrefix(name)
  754. maxResults := int32(1)
  755. pager := fs.containerClient.NewListBlobsFlatPager(&container.ListBlobsFlatOptions{
  756. MaxResults: &maxResults,
  757. Prefix: &prefix,
  758. })
  759. if pager.More() {
  760. ctx, cancelFn := context.WithDeadline(context.Background(), time.Now().Add(fs.ctxTimeout))
  761. defer cancelFn()
  762. resp, err := pager.NextPage(ctx)
  763. if err != nil {
  764. metric.AZListObjectsCompleted(err)
  765. return result, err
  766. }
  767. result = len(resp.ListBlobsFlatSegmentResponse.Segment.BlobItems) > 0
  768. }
  769. metric.AZListObjectsCompleted(nil)
  770. return result, nil
  771. }
  772. func (fs *AzureBlobFs) downloadPart(ctx context.Context, blockBlob *blockblob.Client, buf []byte,
  773. w io.WriterAt, offset, count, writeOffset int64,
  774. ) error {
  775. if count == 0 {
  776. return nil
  777. }
  778. resp, err := blockBlob.DownloadStream(ctx, &blob.DownloadStreamOptions{
  779. Range: blob.HTTPRange{
  780. Offset: offset,
  781. Count: count,
  782. },
  783. })
  784. if err != nil {
  785. return err
  786. }
  787. defer resp.DownloadResponse.Body.Close()
  788. _, err = io.ReadAtLeast(resp.DownloadResponse.Body, buf, int(count))
  789. if err != nil {
  790. return err
  791. }
  792. _, err = fs.writeAtFull(w, buf, writeOffset, int(count))
  793. return err
  794. }
  795. func (fs *AzureBlobFs) handleMultipartDownload(ctx context.Context, blockBlob *blockblob.Client,
  796. offset int64, writer io.WriterAt, pipeReader PipeReader,
  797. ) error {
  798. props, err := blockBlob.GetProperties(ctx, &blob.GetPropertiesOptions{})
  799. metric.AZHeadObjectCompleted(err)
  800. if err != nil {
  801. fsLog(fs, logger.LevelError, "unable to get blob properties, download aborted: %+v", err)
  802. return err
  803. }
  804. if readMetadata > 0 && pipeReader != nil {
  805. pipeReader.setMetadataFromPointerVal(props.Metadata)
  806. }
  807. contentLength := util.GetIntFromPointer(props.ContentLength)
  808. sizeToDownload := contentLength - offset
  809. if sizeToDownload < 0 {
  810. fsLog(fs, logger.LevelError, "invalid multipart download size or offset, size: %v, offset: %v, size to download: %v",
  811. contentLength, offset, sizeToDownload)
  812. return errors.New("the requested offset exceeds the file size")
  813. }
  814. if sizeToDownload == 0 {
  815. fsLog(fs, logger.LevelDebug, "nothing to download, offset %v, content length %v", offset, contentLength)
  816. return nil
  817. }
  818. partSize := fs.config.DownloadPartSize
  819. guard := make(chan struct{}, fs.config.DownloadConcurrency)
  820. blockCtxTimeout := time.Duration(fs.config.DownloadPartSize/(1024*1024)) * time.Minute
  821. pool := newBufferAllocator(int(partSize))
  822. finished := false
  823. var wg sync.WaitGroup
  824. var errOnce sync.Once
  825. var hasError atomic.Bool
  826. var poolError error
  827. poolCtx, poolCancel := context.WithCancel(ctx)
  828. defer poolCancel()
  829. for part := 0; !finished; part++ {
  830. start := offset
  831. end := offset + partSize
  832. if end >= contentLength {
  833. end = contentLength
  834. finished = true
  835. }
  836. writeOffset := int64(part) * partSize
  837. offset = end
  838. guard <- struct{}{}
  839. if hasError.Load() {
  840. fsLog(fs, logger.LevelDebug, "pool error, download for part %v not started", part)
  841. break
  842. }
  843. buf := pool.getBuffer()
  844. wg.Add(1)
  845. go func(start, end, writeOffset int64, buf []byte) {
  846. defer func() {
  847. pool.releaseBuffer(buf)
  848. <-guard
  849. wg.Done()
  850. }()
  851. innerCtx, cancelFn := context.WithDeadline(poolCtx, time.Now().Add(blockCtxTimeout))
  852. defer cancelFn()
  853. count := end - start
  854. err := fs.downloadPart(innerCtx, blockBlob, buf, writer, start, count, writeOffset)
  855. if err != nil {
  856. errOnce.Do(func() {
  857. fsLog(fs, logger.LevelError, "multipart download error: %+v", err)
  858. hasError.Store(true)
  859. poolError = fmt.Errorf("multipart download error: %w", err)
  860. poolCancel()
  861. })
  862. }
  863. }(start, end, writeOffset, buf)
  864. }
  865. wg.Wait()
  866. close(guard)
  867. pool.free()
  868. return poolError
  869. }
  870. func (fs *AzureBlobFs) handleMultipartUpload(ctx context.Context, reader io.Reader,
  871. blockBlob *blockblob.Client, httpHeaders *blob.HTTPHeaders, metadata map[string]*string,
  872. ) error {
  873. partSize := fs.config.UploadPartSize
  874. guard := make(chan struct{}, fs.config.UploadConcurrency)
  875. blockCtxTimeout := time.Duration(fs.config.UploadPartSize/(1024*1024)) * time.Minute
  876. // sync.Pool seems to use a lot of memory so prefer our own, very simple, allocator
  877. // we only need to recycle few byte slices
  878. pool := newBufferAllocator(int(partSize))
  879. finished := false
  880. var blocks []string
  881. var wg sync.WaitGroup
  882. var errOnce sync.Once
  883. var hasError atomic.Bool
  884. var poolError error
  885. poolCtx, poolCancel := context.WithCancel(ctx)
  886. defer poolCancel()
  887. for part := 0; !finished; part++ {
  888. buf := pool.getBuffer()
  889. n, err := fs.readFill(reader, buf)
  890. if err == io.EOF {
  891. // read finished, if n > 0 we need to process the last data chunck
  892. if n == 0 {
  893. pool.releaseBuffer(buf)
  894. break
  895. }
  896. finished = true
  897. } else if err != nil {
  898. pool.releaseBuffer(buf)
  899. pool.free()
  900. return err
  901. }
  902. // Block IDs are unique values to avoid issue if 2+ clients are uploading blocks
  903. // at the same time causing CommitBlockList to get a mix of blocks from all the clients.
  904. generatedUUID, err := uuid.NewRandom()
  905. if err != nil {
  906. pool.releaseBuffer(buf)
  907. pool.free()
  908. return fmt.Errorf("unable to generate block ID: %w", err)
  909. }
  910. blockID := base64.StdEncoding.EncodeToString([]byte(generatedUUID.String()))
  911. blocks = append(blocks, blockID)
  912. guard <- struct{}{}
  913. if hasError.Load() {
  914. fsLog(fs, logger.LevelError, "pool error, upload for part %d not started", part)
  915. pool.releaseBuffer(buf)
  916. break
  917. }
  918. wg.Add(1)
  919. go func(blockID string, buf []byte, bufSize int) {
  920. defer func() {
  921. pool.releaseBuffer(buf)
  922. <-guard
  923. wg.Done()
  924. }()
  925. bufferReader := &bytesReaderWrapper{
  926. Reader: bytes.NewReader(buf[:bufSize]),
  927. }
  928. innerCtx, cancelFn := context.WithDeadline(poolCtx, time.Now().Add(blockCtxTimeout))
  929. defer cancelFn()
  930. _, err := blockBlob.StageBlock(innerCtx, blockID, bufferReader, &blockblob.StageBlockOptions{})
  931. if err != nil {
  932. errOnce.Do(func() {
  933. fsLog(fs, logger.LevelDebug, "multipart upload error: %+v", err)
  934. hasError.Store(true)
  935. poolError = fmt.Errorf("multipart upload error: %w", err)
  936. poolCancel()
  937. })
  938. }
  939. }(blockID, buf, n)
  940. }
  941. wg.Wait()
  942. close(guard)
  943. pool.free()
  944. if poolError != nil {
  945. return poolError
  946. }
  947. commitOptions := blockblob.CommitBlockListOptions{
  948. HTTPHeaders: httpHeaders,
  949. Metadata: metadata,
  950. }
  951. if fs.config.AccessTier != "" {
  952. commitOptions.Tier = (*blob.AccessTier)(&fs.config.AccessTier)
  953. }
  954. _, err := blockBlob.CommitBlockList(ctx, blocks, &commitOptions)
  955. return err
  956. }
  957. func (*AzureBlobFs) writeAtFull(w io.WriterAt, buf []byte, offset int64, count int) (int, error) {
  958. written := 0
  959. for written < count {
  960. n, err := w.WriteAt(buf[written:count], offset+int64(written))
  961. written += n
  962. if err != nil {
  963. return written, err
  964. }
  965. }
  966. return written, nil
  967. }
  968. // copied from rclone
  969. func (*AzureBlobFs) readFill(r io.Reader, buf []byte) (n int, err error) {
  970. var nn int
  971. for n < len(buf) && err == nil {
  972. nn, err = r.Read(buf[n:])
  973. n += nn
  974. }
  975. return n, err
  976. }
  977. func (fs *AzureBlobFs) getCopyOptions() *blob.StartCopyFromURLOptions {
  978. copyOptions := &blob.StartCopyFromURLOptions{}
  979. if fs.config.AccessTier != "" {
  980. copyOptions.Tier = (*blob.AccessTier)(&fs.config.AccessTier)
  981. }
  982. return copyOptions
  983. }
  984. func (fs *AzureBlobFs) downloadToWriter(name string, w PipeWriter) (int64, error) {
  985. fsLog(fs, logger.LevelDebug, "starting download before resuming upload, path %q", name)
  986. ctx, cancelFn := context.WithTimeout(context.Background(), preResumeTimeout)
  987. defer cancelFn()
  988. blockBlob := fs.containerClient.NewBlockBlobClient(name)
  989. err := fs.handleMultipartDownload(ctx, blockBlob, 0, w, nil)
  990. n := w.GetWrittenBytes()
  991. fsLog(fs, logger.LevelDebug, "download before resuming upload completed, path %q size: %d, err: %+v",
  992. name, n, err)
  993. metric.AZTransferCompleted(n, 1, err)
  994. return n, err
  995. }
  996. func checkDirectoryMarkers(contentType string, metadata map[string]*string) bool {
  997. if contentType == dirMimeType {
  998. return true
  999. }
  1000. for k, v := range metadata {
  1001. if strings.ToLower(k) == azFolderKey {
  1002. return strings.ToLower(util.GetStringFromPointer(v)) == "true"
  1003. }
  1004. }
  1005. return false
  1006. }
  1007. func getAzContainerClientOptions() *container.ClientOptions {
  1008. version := version.Get()
  1009. return &container.ClientOptions{
  1010. ClientOptions: azcore.ClientOptions{
  1011. Telemetry: policy.TelemetryOptions{
  1012. ApplicationID: fmt.Sprintf("SFTPGo-%s", version.CommitHash),
  1013. },
  1014. },
  1015. }
  1016. }
  1017. type bytesReaderWrapper struct {
  1018. *bytes.Reader
  1019. }
  1020. func (b *bytesReaderWrapper) Close() error {
  1021. return nil
  1022. }
  1023. type bufferAllocator struct {
  1024. sync.Mutex
  1025. available [][]byte
  1026. bufferSize int
  1027. finalized bool
  1028. }
  1029. func newBufferAllocator(size int) *bufferAllocator {
  1030. return &bufferAllocator{
  1031. bufferSize: size,
  1032. finalized: false,
  1033. }
  1034. }
  1035. func (b *bufferAllocator) getBuffer() []byte {
  1036. b.Lock()
  1037. defer b.Unlock()
  1038. if len(b.available) > 0 {
  1039. var result []byte
  1040. truncLength := len(b.available) - 1
  1041. result = b.available[truncLength]
  1042. b.available[truncLength] = nil
  1043. b.available = b.available[:truncLength]
  1044. return result
  1045. }
  1046. return make([]byte, b.bufferSize)
  1047. }
  1048. func (b *bufferAllocator) releaseBuffer(buf []byte) {
  1049. b.Lock()
  1050. defer b.Unlock()
  1051. if b.finalized || len(buf) != b.bufferSize {
  1052. return
  1053. }
  1054. b.available = append(b.available, buf)
  1055. }
  1056. func (b *bufferAllocator) free() {
  1057. b.Lock()
  1058. defer b.Unlock()
  1059. b.available = nil
  1060. b.finalized = true
  1061. }
  1062. type azureBlobDirLister struct {
  1063. baseDirLister
  1064. paginator *runtime.Pager[container.ListBlobsHierarchyResponse]
  1065. timeout time.Duration
  1066. prefix string
  1067. prefixes map[string]bool
  1068. metricUpdated bool
  1069. }
  1070. func (l *azureBlobDirLister) Next(limit int) ([]os.FileInfo, error) {
  1071. if limit <= 0 {
  1072. return nil, errInvalidDirListerLimit
  1073. }
  1074. if len(l.cache) >= limit {
  1075. return l.returnFromCache(limit), nil
  1076. }
  1077. if !l.paginator.More() {
  1078. if !l.metricUpdated {
  1079. l.metricUpdated = true
  1080. metric.AZListObjectsCompleted(nil)
  1081. }
  1082. return l.returnFromCache(limit), io.EOF
  1083. }
  1084. ctx, cancelFn := context.WithDeadline(context.Background(), time.Now().Add(l.timeout))
  1085. defer cancelFn()
  1086. page, err := l.paginator.NextPage(ctx)
  1087. if err != nil {
  1088. metric.AZListObjectsCompleted(err)
  1089. return l.cache, err
  1090. }
  1091. for _, blobPrefix := range page.ListBlobsHierarchySegmentResponse.Segment.BlobPrefixes {
  1092. name := util.GetStringFromPointer(blobPrefix.Name)
  1093. // we don't support prefixes == "/" this will be sent if a key starts with "/"
  1094. if name == "" || name == "/" {
  1095. continue
  1096. }
  1097. // sometime we have duplicate prefixes, maybe an Azurite bug
  1098. name = strings.TrimPrefix(name, l.prefix)
  1099. if _, ok := l.prefixes[strings.TrimSuffix(name, "/")]; ok {
  1100. continue
  1101. }
  1102. l.cache = append(l.cache, NewFileInfo(name, true, 0, time.Unix(0, 0), false))
  1103. l.prefixes[strings.TrimSuffix(name, "/")] = true
  1104. }
  1105. for _, blobItem := range page.ListBlobsHierarchySegmentResponse.Segment.BlobItems {
  1106. name := util.GetStringFromPointer(blobItem.Name)
  1107. name = strings.TrimPrefix(name, l.prefix)
  1108. size := int64(0)
  1109. isDir := false
  1110. modTime := time.Unix(0, 0)
  1111. if blobItem.Properties != nil {
  1112. size = util.GetIntFromPointer(blobItem.Properties.ContentLength)
  1113. modTime = util.GetTimeFromPointer(blobItem.Properties.LastModified)
  1114. contentType := util.GetStringFromPointer(blobItem.Properties.ContentType)
  1115. isDir = checkDirectoryMarkers(contentType, blobItem.Metadata)
  1116. if isDir {
  1117. // check if the dir is already included, it will be sent as blob prefix if it contains at least one item
  1118. if _, ok := l.prefixes[name]; ok {
  1119. continue
  1120. }
  1121. l.prefixes[name] = true
  1122. }
  1123. if val := getAzureLastModified(blobItem.Metadata); val > 0 {
  1124. modTime = util.GetTimeFromMsecSinceEpoch(val)
  1125. }
  1126. }
  1127. l.cache = append(l.cache, NewFileInfo(name, isDir, size, modTime, false))
  1128. }
  1129. return l.returnFromCache(limit), nil
  1130. }
  1131. func (l *azureBlobDirLister) Close() error {
  1132. clear(l.prefixes)
  1133. return l.baseDirLister.Close()
  1134. }