gcsfs.go 24 KB

123456789101112131415161718192021222324252627282930313233343536373839404142434445464748495051525354555657585960616263646566676869707172737475767778798081828384858687888990919293949596979899100101102103104105106107108109110111112113114115116117118119120121122123124125126127128129130131132133134135136137138139140141142143144145146147148149150151152153154155156157158159160161162163164165166167168169170171172173174175176177178179180181182183184185186187188189190191192193194195196197198199200201202203204205206207208209210211212213214215216217218219220221222223224225226227228229230231232233234235236237238239240241242243244245246247248249250251252253254255256257258259260261262263264265266267268269270271272273274275276277278279280281282283284285286287288289290291292293294295296297298299300301302303304305306307308309310311312313314315316317318319320321322323324325326327328329330331332333334335336337338339340341342343344345346347348349350351352353354355356357358359360361362363364365366367368369370371372373374375376377378379380381382383384385386387388389390391392393394395396397398399400401402403404405406407408409410411412413414415416417418419420421422423424425426427428429430431432433434435436437438439440441442443444445446447448449450451452453454455456457458459460461462463464465466467468469470471472473474475476477478479480481482483484485486487488489490491492493494495496497498499500501502503504505506507508509510511512513514515516517518519520521522523524525526527528529530531532533534535536537538539540541542543544545546547548549550551552553554555556557558559560561562563564565566567568569570571572573574575576577578579580581582583584585586587588589590591592593594595596597598599600601602603604605606607608609610611612613614615616617618619620621622623624625626627628629630631632633634635636637638639640641642643644645646647648649650651652653654655656657658659660661662663664665666667668669670671672673674675676677678679680681682683684685686687688689690691692693694695696697698699700701702703704705706707708709710711712713714715716717718719720721722723724725726727728729730731732733734735736737738739740741742743744745746747748749750751752753754755756757758759760761762763764765766767768769770771772773774775776777778779780781782783784785786787788789790791792793794795796797798799800801802803804805806807808809810811812813814815816817818819820821822823824825826827828829830831832833834835836837838839840841842843844845846847848849850851852853854855856857858859860861862863864865866867868869870871872873874875876877878879880881882883884885886
  1. // Copyright (C) 2019-2022 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 !nogcs
  15. // +build !nogcs
  16. package vfs
  17. import (
  18. "context"
  19. "fmt"
  20. "io"
  21. "mime"
  22. "net/http"
  23. "os"
  24. "path"
  25. "path/filepath"
  26. "strings"
  27. "time"
  28. "cloud.google.com/go/storage"
  29. "github.com/eikenb/pipeat"
  30. "github.com/pkg/sftp"
  31. "google.golang.org/api/googleapi"
  32. "google.golang.org/api/iterator"
  33. "google.golang.org/api/option"
  34. "github.com/drakkan/sftpgo/v2/internal/logger"
  35. "github.com/drakkan/sftpgo/v2/internal/metric"
  36. "github.com/drakkan/sftpgo/v2/internal/plugin"
  37. "github.com/drakkan/sftpgo/v2/internal/util"
  38. "github.com/drakkan/sftpgo/v2/internal/version"
  39. )
  40. const (
  41. defaultGCSPageSize = 5000
  42. )
  43. var (
  44. gcsDefaultFieldsSelection = []string{"Name", "Size", "Deleted", "Updated", "ContentType"}
  45. )
  46. // GCSFs is a Fs implementation for Google Cloud Storage.
  47. type GCSFs struct {
  48. connectionID string
  49. localTempDir string
  50. // if not empty this fs is mouted as virtual folder in the specified path
  51. mountPath string
  52. config *GCSFsConfig
  53. svc *storage.Client
  54. ctxTimeout time.Duration
  55. ctxLongTimeout time.Duration
  56. }
  57. func init() {
  58. version.AddFeature("+gcs")
  59. }
  60. // NewGCSFs returns an GCSFs object that allows to interact with Google Cloud Storage
  61. func NewGCSFs(connectionID, localTempDir, mountPath string, config GCSFsConfig) (Fs, error) {
  62. if localTempDir == "" {
  63. if tempPath != "" {
  64. localTempDir = tempPath
  65. } else {
  66. localTempDir = filepath.Clean(os.TempDir())
  67. }
  68. }
  69. var err error
  70. fs := &GCSFs{
  71. connectionID: connectionID,
  72. localTempDir: localTempDir,
  73. mountPath: getMountPath(mountPath),
  74. config: &config,
  75. ctxTimeout: 30 * time.Second,
  76. ctxLongTimeout: 300 * time.Second,
  77. }
  78. if err = fs.config.validate(); err != nil {
  79. return fs, err
  80. }
  81. ctx := context.Background()
  82. if fs.config.AutomaticCredentials > 0 {
  83. fs.svc, err = storage.NewClient(ctx)
  84. } else {
  85. err = fs.config.Credentials.TryDecrypt()
  86. if err != nil {
  87. return fs, err
  88. }
  89. fs.svc, err = storage.NewClient(ctx, option.WithCredentialsJSON([]byte(fs.config.Credentials.GetPayload())))
  90. }
  91. return fs, err
  92. }
  93. // Name returns the name for the Fs implementation
  94. func (fs *GCSFs) Name() string {
  95. return fmt.Sprintf("%s bucket %q", gcsfsName, fs.config.Bucket)
  96. }
  97. // ConnectionID returns the connection ID associated to this Fs implementation
  98. func (fs *GCSFs) ConnectionID() string {
  99. return fs.connectionID
  100. }
  101. // Stat returns a FileInfo describing the named file
  102. func (fs *GCSFs) Stat(name string) (os.FileInfo, error) {
  103. if name == "" || name == "/" || name == "." {
  104. return updateFileInfoModTime(fs.getStorageID(), name, NewFileInfo(name, true, 0, time.Unix(0, 0), false))
  105. }
  106. if fs.config.KeyPrefix == name+"/" {
  107. return updateFileInfoModTime(fs.getStorageID(), name, NewFileInfo(name, true, 0, time.Unix(0, 0), false))
  108. }
  109. _, info, err := fs.getObjectStat(name)
  110. return info, err
  111. }
  112. // Lstat returns a FileInfo describing the named file
  113. func (fs *GCSFs) Lstat(name string) (os.FileInfo, error) {
  114. return fs.Stat(name)
  115. }
  116. // Open opens the named file for reading
  117. func (fs *GCSFs) Open(name string, offset int64) (File, *pipeat.PipeReaderAt, func(), error) {
  118. r, w, err := pipeat.PipeInDir(fs.localTempDir)
  119. if err != nil {
  120. return nil, nil, nil, err
  121. }
  122. bkt := fs.svc.Bucket(fs.config.Bucket)
  123. obj := bkt.Object(name)
  124. ctx, cancelFn := context.WithCancel(context.Background())
  125. objectReader, err := obj.NewRangeReader(ctx, offset, -1)
  126. if err == nil && offset > 0 && objectReader.Attrs.ContentEncoding == "gzip" {
  127. err = fmt.Errorf("range request is not possible for gzip content encoding, requested offset %v", offset)
  128. objectReader.Close()
  129. }
  130. if err != nil {
  131. r.Close()
  132. w.Close()
  133. cancelFn()
  134. return nil, nil, nil, err
  135. }
  136. go func() {
  137. defer cancelFn()
  138. defer objectReader.Close()
  139. n, err := io.Copy(w, objectReader)
  140. w.CloseWithError(err) //nolint:errcheck
  141. fsLog(fs, logger.LevelDebug, "download completed, path: %#v size: %v, err: %+v", name, n, err)
  142. metric.GCSTransferCompleted(n, 1, err)
  143. }()
  144. return nil, r, cancelFn, nil
  145. }
  146. // Create creates or opens the named file for writing
  147. func (fs *GCSFs) Create(name string, flag, checks int) (File, *PipeWriter, func(), error) {
  148. if checks&CheckParentDir != 0 {
  149. _, err := fs.Stat(path.Dir(name))
  150. if err != nil {
  151. return nil, nil, nil, err
  152. }
  153. }
  154. r, w, err := pipeat.PipeInDir(fs.localTempDir)
  155. if err != nil {
  156. return nil, nil, nil, err
  157. }
  158. p := NewPipeWriter(w)
  159. bkt := fs.svc.Bucket(fs.config.Bucket)
  160. obj := bkt.Object(name)
  161. if flag == -1 {
  162. obj = obj.If(storage.Conditions{DoesNotExist: true})
  163. } else {
  164. attrs, statErr := fs.headObject(name)
  165. if statErr == nil {
  166. obj = obj.If(storage.Conditions{GenerationMatch: attrs.Generation})
  167. } else if fs.IsNotExist(statErr) {
  168. obj = obj.If(storage.Conditions{DoesNotExist: true})
  169. } else {
  170. fsLog(fs, logger.LevelWarn, "unable to set precondition for %q, stat err: %v", name, statErr)
  171. }
  172. }
  173. ctx, cancelFn := context.WithCancel(context.Background())
  174. objectWriter := obj.NewWriter(ctx)
  175. var contentType string
  176. if flag == -1 {
  177. contentType = dirMimeType
  178. } else {
  179. contentType = mime.TypeByExtension(path.Ext(name))
  180. }
  181. if contentType != "" {
  182. objectWriter.ObjectAttrs.ContentType = contentType
  183. }
  184. if fs.config.StorageClass != "" {
  185. objectWriter.ObjectAttrs.StorageClass = fs.config.StorageClass
  186. }
  187. if fs.config.ACL != "" {
  188. objectWriter.PredefinedACL = fs.config.ACL
  189. }
  190. go func() {
  191. defer cancelFn()
  192. n, err := io.Copy(objectWriter, r)
  193. closeErr := objectWriter.Close()
  194. if err == nil {
  195. err = closeErr
  196. }
  197. r.CloseWithError(err) //nolint:errcheck
  198. p.Done(err)
  199. fsLog(fs, logger.LevelDebug, "upload completed, path: %#v, acl: %#v, readed bytes: %v, err: %+v",
  200. name, fs.config.ACL, n, err)
  201. metric.GCSTransferCompleted(n, 0, err)
  202. }()
  203. return nil, p, cancelFn, nil
  204. }
  205. // Rename renames (moves) source to target.
  206. // We don't support renaming non empty directories since we should
  207. // rename all the contents too and this could take long time: think
  208. // about directories with thousands of files, for each file we should
  209. // execute a CopyObject call.
  210. func (fs *GCSFs) Rename(source, target string) error { //nolint:gocyclo
  211. if source == target {
  212. return nil
  213. }
  214. _, err := fs.Stat(path.Dir(target))
  215. if err != nil {
  216. return err
  217. }
  218. realSourceName, fi, err := fs.getObjectStat(source)
  219. if err != nil {
  220. return err
  221. }
  222. if fi.IsDir() {
  223. hasContents, err := fs.hasContents(source)
  224. if err != nil {
  225. return err
  226. }
  227. if hasContents {
  228. return fmt.Errorf("cannot rename non empty directory: %#v", source)
  229. }
  230. if err := fs.mkdirInternal(target); err != nil {
  231. return err
  232. }
  233. } else {
  234. src := fs.svc.Bucket(fs.config.Bucket).Object(realSourceName)
  235. dst := fs.svc.Bucket(fs.config.Bucket).Object(target)
  236. attrs, statErr := fs.headObject(target)
  237. if statErr == nil {
  238. dst = dst.If(storage.Conditions{GenerationMatch: attrs.Generation})
  239. } else if fs.IsNotExist(statErr) {
  240. dst = dst.If(storage.Conditions{DoesNotExist: true})
  241. } else {
  242. fsLog(fs, logger.LevelWarn, "unable to set precondition for rename, target %q, stat err: %v",
  243. target, statErr)
  244. }
  245. ctx, cancelFn := context.WithDeadline(context.Background(), time.Now().Add(fs.ctxTimeout))
  246. defer cancelFn()
  247. copier := dst.CopierFrom(src)
  248. if fs.config.StorageClass != "" {
  249. copier.StorageClass = fs.config.StorageClass
  250. }
  251. if fs.config.ACL != "" {
  252. copier.PredefinedACL = fs.config.ACL
  253. }
  254. contentType := mime.TypeByExtension(path.Ext(source))
  255. if contentType != "" {
  256. copier.ContentType = contentType
  257. }
  258. _, err = copier.Run(ctx)
  259. metric.GCSCopyObjectCompleted(err)
  260. if err != nil {
  261. return err
  262. }
  263. if plugin.Handler.HasMetadater() {
  264. err = plugin.Handler.SetModificationTime(fs.getStorageID(), ensureAbsPath(target),
  265. util.GetTimeAsMsSinceEpoch(fi.ModTime()))
  266. if err != nil {
  267. fsLog(fs, logger.LevelWarn, "unable to preserve modification time after renaming %#v -> %#v: %+v",
  268. source, target, err)
  269. }
  270. }
  271. }
  272. return fs.Remove(source, fi.IsDir())
  273. }
  274. // Remove removes the named file or (empty) directory.
  275. func (fs *GCSFs) Remove(name string, isDir bool) error {
  276. if isDir {
  277. hasContents, err := fs.hasContents(name)
  278. if err != nil {
  279. return err
  280. }
  281. if hasContents {
  282. return fmt.Errorf("cannot remove non empty directory: %#v", name)
  283. }
  284. if !strings.HasSuffix(name, "/") {
  285. name += "/"
  286. }
  287. }
  288. obj := fs.svc.Bucket(fs.config.Bucket).Object(name)
  289. attrs, statErr := fs.headObject(name)
  290. if statErr == nil {
  291. obj = obj.If(storage.Conditions{GenerationMatch: attrs.Generation})
  292. } else {
  293. fsLog(fs, logger.LevelWarn, "unable to set precondition for deleting %q, stat err: %v",
  294. name, statErr)
  295. }
  296. ctx, cancelFn := context.WithDeadline(context.Background(), time.Now().Add(fs.ctxTimeout))
  297. defer cancelFn()
  298. err := obj.Delete(ctx)
  299. if isDir && fs.IsNotExist(err) {
  300. // we can have directories without a trailing "/" (created using v2.1.0 and before)
  301. err = fs.svc.Bucket(fs.config.Bucket).Object(strings.TrimSuffix(name, "/")).Delete(ctx)
  302. }
  303. metric.GCSDeleteObjectCompleted(err)
  304. if plugin.Handler.HasMetadater() && err == nil && !isDir {
  305. if errMetadata := plugin.Handler.RemoveMetadata(fs.getStorageID(), ensureAbsPath(name)); errMetadata != nil {
  306. fsLog(fs, logger.LevelWarn, "unable to remove metadata for path %#v: %+v", name, errMetadata)
  307. }
  308. }
  309. return err
  310. }
  311. // Mkdir creates a new directory with the specified name and default permissions
  312. func (fs *GCSFs) Mkdir(name string) error {
  313. _, err := fs.Stat(name)
  314. if !fs.IsNotExist(err) {
  315. return err
  316. }
  317. return fs.mkdirInternal(name)
  318. }
  319. // Symlink creates source as a symbolic link to target.
  320. func (*GCSFs) Symlink(source, target string) error {
  321. return ErrVfsUnsupported
  322. }
  323. // Readlink returns the destination of the named symbolic link
  324. func (*GCSFs) Readlink(name string) (string, error) {
  325. return "", ErrVfsUnsupported
  326. }
  327. // Chown changes the numeric uid and gid of the named file.
  328. func (*GCSFs) Chown(name string, uid int, gid int) error {
  329. return ErrVfsUnsupported
  330. }
  331. // Chmod changes the mode of the named file to mode.
  332. func (*GCSFs) Chmod(name string, mode os.FileMode) error {
  333. return ErrVfsUnsupported
  334. }
  335. // Chtimes changes the access and modification times of the named file.
  336. func (fs *GCSFs) Chtimes(name string, atime, mtime time.Time, isUploading bool) error {
  337. if !plugin.Handler.HasMetadater() {
  338. return ErrVfsUnsupported
  339. }
  340. if !isUploading {
  341. info, err := fs.Stat(name)
  342. if err != nil {
  343. return err
  344. }
  345. if info.IsDir() {
  346. return ErrVfsUnsupported
  347. }
  348. }
  349. return plugin.Handler.SetModificationTime(fs.getStorageID(), ensureAbsPath(name),
  350. util.GetTimeAsMsSinceEpoch(mtime))
  351. }
  352. // Truncate changes the size of the named file.
  353. // Truncate by path is not supported, while truncating an opened
  354. // file is handled inside base transfer
  355. func (*GCSFs) Truncate(name string, size int64) error {
  356. return ErrVfsUnsupported
  357. }
  358. // ReadDir reads the directory named by dirname and returns
  359. // a list of directory entries.
  360. func (fs *GCSFs) ReadDir(dirname string) ([]os.FileInfo, error) {
  361. var result []os.FileInfo
  362. // dirname must be already cleaned
  363. prefix := fs.getPrefix(dirname)
  364. query := &storage.Query{Prefix: prefix, Delimiter: "/"}
  365. err := query.SetAttrSelection(gcsDefaultFieldsSelection)
  366. if err != nil {
  367. return nil, err
  368. }
  369. modTimes, err := getFolderModTimes(fs.getStorageID(), dirname)
  370. if err != nil {
  371. return result, err
  372. }
  373. prefixes := make(map[string]bool)
  374. ctx, cancelFn := context.WithDeadline(context.Background(), time.Now().Add(fs.ctxLongTimeout))
  375. defer cancelFn()
  376. bkt := fs.svc.Bucket(fs.config.Bucket)
  377. it := bkt.Objects(ctx, query)
  378. pager := iterator.NewPager(it, defaultGCSPageSize, "")
  379. for {
  380. var objects []*storage.ObjectAttrs
  381. pageToken, err := pager.NextPage(&objects)
  382. if err != nil {
  383. metric.GCSListObjectsCompleted(err)
  384. return result, err
  385. }
  386. for _, attrs := range objects {
  387. if attrs.Prefix != "" {
  388. name, _ := fs.resolve(attrs.Prefix, prefix, attrs.ContentType)
  389. if name == "" {
  390. continue
  391. }
  392. if _, ok := prefixes[name]; ok {
  393. continue
  394. }
  395. result = append(result, NewFileInfo(name, true, 0, time.Unix(0, 0), false))
  396. prefixes[name] = true
  397. } else {
  398. name, isDir := fs.resolve(attrs.Name, prefix, attrs.ContentType)
  399. if name == "" {
  400. continue
  401. }
  402. if !attrs.Deleted.IsZero() {
  403. continue
  404. }
  405. if isDir {
  406. // check if the dir is already included, it will be sent as blob prefix if it contains at least one item
  407. if _, ok := prefixes[name]; ok {
  408. continue
  409. }
  410. prefixes[name] = true
  411. }
  412. modTime := attrs.Updated
  413. if t, ok := modTimes[name]; ok {
  414. modTime = util.GetTimeFromMsecSinceEpoch(t)
  415. }
  416. result = append(result, NewFileInfo(name, isDir, attrs.Size, modTime, false))
  417. }
  418. }
  419. objects = nil
  420. if pageToken == "" {
  421. break
  422. }
  423. }
  424. metric.GCSListObjectsCompleted(nil)
  425. return result, nil
  426. }
  427. // IsUploadResumeSupported returns true if resuming uploads is supported.
  428. // Resuming uploads is not supported on GCS
  429. func (*GCSFs) IsUploadResumeSupported() bool {
  430. return false
  431. }
  432. // IsAtomicUploadSupported returns true if atomic upload is supported.
  433. // S3 uploads are already atomic, we don't need to upload to a temporary
  434. // file
  435. func (*GCSFs) IsAtomicUploadSupported() bool {
  436. return false
  437. }
  438. // IsNotExist returns a boolean indicating whether the error is known to
  439. // report that a file or directory does not exist
  440. func (*GCSFs) IsNotExist(err error) bool {
  441. if err == nil {
  442. return false
  443. }
  444. if err == storage.ErrObjectNotExist || err == storage.ErrBucketNotExist {
  445. return true
  446. }
  447. if e, ok := err.(*googleapi.Error); ok {
  448. if e.Code == http.StatusNotFound {
  449. return true
  450. }
  451. }
  452. return false
  453. }
  454. // IsPermission returns a boolean indicating whether the error is known to
  455. // report that permission is denied.
  456. func (*GCSFs) IsPermission(err error) bool {
  457. if err == nil {
  458. return false
  459. }
  460. if e, ok := err.(*googleapi.Error); ok {
  461. if e.Code == http.StatusForbidden || e.Code == http.StatusUnauthorized {
  462. return true
  463. }
  464. }
  465. return false
  466. }
  467. // IsNotSupported returns true if the error indicate an unsupported operation
  468. func (*GCSFs) IsNotSupported(err error) bool {
  469. if err == nil {
  470. return false
  471. }
  472. return err == ErrVfsUnsupported
  473. }
  474. // CheckRootPath creates the specified local root directory if it does not exists
  475. func (fs *GCSFs) CheckRootPath(username string, uid int, gid int) bool {
  476. // we need a local directory for temporary files
  477. osFs := NewOsFs(fs.ConnectionID(), fs.localTempDir, "")
  478. return osFs.CheckRootPath(username, uid, gid)
  479. }
  480. // ScanRootDirContents returns the number of files contained in the bucket,
  481. // and their size
  482. func (fs *GCSFs) ScanRootDirContents() (int, int64, error) {
  483. return fs.GetDirSize(fs.config.KeyPrefix)
  484. }
  485. func (fs *GCSFs) getFileNamesInPrefix(fsPrefix string) (map[string]bool, error) {
  486. fileNames := make(map[string]bool)
  487. prefix := ""
  488. if fsPrefix != "/" {
  489. prefix = strings.TrimPrefix(fsPrefix, "/")
  490. }
  491. query := &storage.Query{
  492. Prefix: prefix,
  493. Delimiter: "/",
  494. }
  495. err := query.SetAttrSelection(gcsDefaultFieldsSelection)
  496. if err != nil {
  497. return fileNames, err
  498. }
  499. ctx, cancelFn := context.WithDeadline(context.Background(), time.Now().Add(fs.ctxLongTimeout))
  500. defer cancelFn()
  501. bkt := fs.svc.Bucket(fs.config.Bucket)
  502. it := bkt.Objects(ctx, query)
  503. pager := iterator.NewPager(it, defaultGCSPageSize, "")
  504. for {
  505. var objects []*storage.ObjectAttrs
  506. pageToken, err := pager.NextPage(&objects)
  507. if err != nil {
  508. metric.GCSListObjectsCompleted(err)
  509. return fileNames, err
  510. }
  511. for _, attrs := range objects {
  512. if !attrs.Deleted.IsZero() {
  513. continue
  514. }
  515. if attrs.Prefix == "" {
  516. name, isDir := fs.resolve(attrs.Name, prefix, attrs.ContentType)
  517. if name == "" {
  518. continue
  519. }
  520. if isDir {
  521. continue
  522. }
  523. fileNames[name] = true
  524. }
  525. }
  526. objects = nil
  527. if pageToken == "" {
  528. break
  529. }
  530. }
  531. metric.GCSListObjectsCompleted(nil)
  532. return fileNames, nil
  533. }
  534. // CheckMetadata checks the metadata consistency
  535. func (fs *GCSFs) CheckMetadata() error {
  536. return fsMetadataCheck(fs, fs.getStorageID(), fs.config.KeyPrefix)
  537. }
  538. // GetDirSize returns the number of files and the size for a folder
  539. // including any subfolders
  540. func (fs *GCSFs) GetDirSize(dirname string) (int, int64, error) {
  541. prefix := fs.getPrefix(dirname)
  542. numFiles := 0
  543. size := int64(0)
  544. query := &storage.Query{Prefix: prefix}
  545. err := query.SetAttrSelection(gcsDefaultFieldsSelection)
  546. if err != nil {
  547. return numFiles, size, err
  548. }
  549. ctx, cancelFn := context.WithDeadline(context.Background(), time.Now().Add(fs.ctxLongTimeout))
  550. defer cancelFn()
  551. bkt := fs.svc.Bucket(fs.config.Bucket)
  552. it := bkt.Objects(ctx, query)
  553. pager := iterator.NewPager(it, defaultGCSPageSize, "")
  554. for {
  555. var objects []*storage.ObjectAttrs
  556. pageToken, err := pager.NextPage(&objects)
  557. if err != nil {
  558. metric.GCSListObjectsCompleted(err)
  559. return numFiles, size, err
  560. }
  561. for _, attrs := range objects {
  562. if !attrs.Deleted.IsZero() {
  563. continue
  564. }
  565. isDir := strings.HasSuffix(attrs.Name, "/") || attrs.ContentType == dirMimeType
  566. if isDir && attrs.Size == 0 {
  567. continue
  568. }
  569. numFiles++
  570. size += attrs.Size
  571. if numFiles%1000 == 0 {
  572. fsLog(fs, logger.LevelDebug, "dirname %q scan in progress, files: %d, size: %d", dirname, numFiles, size)
  573. }
  574. }
  575. objects = nil
  576. if pageToken == "" {
  577. break
  578. }
  579. }
  580. metric.GCSListObjectsCompleted(nil)
  581. return numFiles, size, err
  582. }
  583. // GetAtomicUploadPath returns the path to use for an atomic upload.
  584. // GCS uploads are already atomic, we never call this method for GCS
  585. func (*GCSFs) GetAtomicUploadPath(name string) string {
  586. return ""
  587. }
  588. // GetRelativePath returns the path for a file relative to the user's home dir.
  589. // This is the path as seen by SFTPGo users
  590. func (fs *GCSFs) GetRelativePath(name string) string {
  591. rel := path.Clean(name)
  592. if rel == "." {
  593. rel = ""
  594. }
  595. if !path.IsAbs(rel) {
  596. rel = "/" + rel
  597. }
  598. if fs.config.KeyPrefix != "" {
  599. if !strings.HasPrefix(rel, "/"+fs.config.KeyPrefix) {
  600. rel = "/"
  601. }
  602. rel = path.Clean("/" + strings.TrimPrefix(rel, "/"+fs.config.KeyPrefix))
  603. }
  604. if fs.mountPath != "" {
  605. rel = path.Join(fs.mountPath, rel)
  606. }
  607. return rel
  608. }
  609. // Walk walks the file tree rooted at root, calling walkFn for each file or
  610. // directory in the tree, including root
  611. func (fs *GCSFs) Walk(root string, walkFn filepath.WalkFunc) error {
  612. prefix := fs.getPrefix(root)
  613. query := &storage.Query{Prefix: prefix}
  614. err := query.SetAttrSelection(gcsDefaultFieldsSelection)
  615. if err != nil {
  616. walkFn(root, nil, err) //nolint:errcheck
  617. return err
  618. }
  619. ctx, cancelFn := context.WithDeadline(context.Background(), time.Now().Add(fs.ctxLongTimeout))
  620. defer cancelFn()
  621. bkt := fs.svc.Bucket(fs.config.Bucket)
  622. it := bkt.Objects(ctx, query)
  623. pager := iterator.NewPager(it, defaultGCSPageSize, "")
  624. for {
  625. var objects []*storage.ObjectAttrs
  626. pageToken, err := pager.NextPage(&objects)
  627. if err != nil {
  628. walkFn(root, nil, err) //nolint:errcheck
  629. metric.GCSListObjectsCompleted(err)
  630. return err
  631. }
  632. for _, attrs := range objects {
  633. if !attrs.Deleted.IsZero() {
  634. continue
  635. }
  636. name, isDir := fs.resolve(attrs.Name, prefix, attrs.ContentType)
  637. if name == "" {
  638. continue
  639. }
  640. err = walkFn(attrs.Name, NewFileInfo(name, isDir, attrs.Size, attrs.Updated, false), nil)
  641. if err != nil {
  642. return err
  643. }
  644. }
  645. objects = nil
  646. if pageToken == "" {
  647. break
  648. }
  649. }
  650. walkFn(root, NewFileInfo(root, true, 0, time.Unix(0, 0), false), err) //nolint:errcheck
  651. metric.GCSListObjectsCompleted(err)
  652. return err
  653. }
  654. // Join joins any number of path elements into a single path
  655. func (*GCSFs) Join(elem ...string) string {
  656. return strings.TrimPrefix(path.Join(elem...), "/")
  657. }
  658. // HasVirtualFolders returns true if folders are emulated
  659. func (GCSFs) HasVirtualFolders() bool {
  660. return true
  661. }
  662. // ResolvePath returns the matching filesystem path for the specified virtual path
  663. func (fs *GCSFs) ResolvePath(virtualPath string) (string, error) {
  664. if fs.mountPath != "" {
  665. virtualPath = strings.TrimPrefix(virtualPath, fs.mountPath)
  666. }
  667. if !path.IsAbs(virtualPath) {
  668. virtualPath = path.Clean("/" + virtualPath)
  669. }
  670. return fs.Join(fs.config.KeyPrefix, strings.TrimPrefix(virtualPath, "/")), nil
  671. }
  672. func (fs *GCSFs) resolve(name, prefix, contentType string) (string, bool) {
  673. result := strings.TrimPrefix(name, prefix)
  674. isDir := strings.HasSuffix(result, "/")
  675. if isDir {
  676. result = strings.TrimSuffix(result, "/")
  677. }
  678. if contentType == dirMimeType {
  679. isDir = true
  680. }
  681. return result, isDir
  682. }
  683. // getObjectStat returns the stat result and the real object name as first value
  684. func (fs *GCSFs) getObjectStat(name string) (string, os.FileInfo, error) {
  685. attrs, err := fs.headObject(name)
  686. var info os.FileInfo
  687. if err == nil {
  688. objSize := attrs.Size
  689. objectModTime := attrs.Updated
  690. isDir := attrs.ContentType == dirMimeType || strings.HasSuffix(attrs.Name, "/")
  691. info, err = updateFileInfoModTime(fs.getStorageID(), name, NewFileInfo(name, isDir, objSize, objectModTime, false))
  692. return name, info, err
  693. }
  694. if !fs.IsNotExist(err) {
  695. return "", nil, err
  696. }
  697. // now check if this is a prefix (virtual directory)
  698. hasContents, err := fs.hasContents(name)
  699. if err != nil {
  700. return "", nil, err
  701. }
  702. if hasContents {
  703. info, err = updateFileInfoModTime(fs.getStorageID(), name, NewFileInfo(name, true, 0, time.Unix(0, 0), false))
  704. return name, info, err
  705. }
  706. // finally check if this is an object with a trailing /
  707. attrs, err = fs.headObject(name + "/")
  708. if err != nil {
  709. return "", nil, err
  710. }
  711. info, err = updateFileInfoModTime(fs.getStorageID(), name, NewFileInfo(name, true, attrs.Size, attrs.Updated, false))
  712. return name + "/", info, err
  713. }
  714. func (fs *GCSFs) mkdirInternal(name string) error {
  715. if !strings.HasSuffix(name, "/") {
  716. name += "/"
  717. }
  718. _, w, _, err := fs.Create(name, -1, 0)
  719. if err != nil {
  720. return err
  721. }
  722. return w.Close()
  723. }
  724. func (fs *GCSFs) hasContents(name string) (bool, error) {
  725. result := false
  726. prefix := fs.getPrefix(name)
  727. query := &storage.Query{Prefix: prefix}
  728. err := query.SetAttrSelection(gcsDefaultFieldsSelection)
  729. if err != nil {
  730. return result, err
  731. }
  732. ctx, cancelFn := context.WithDeadline(context.Background(), time.Now().Add(fs.ctxTimeout))
  733. defer cancelFn()
  734. bkt := fs.svc.Bucket(fs.config.Bucket)
  735. it := bkt.Objects(ctx, query)
  736. // if we have a dir object with a trailing slash it will be returned so we set the size to 2
  737. pager := iterator.NewPager(it, 2, "")
  738. var objects []*storage.ObjectAttrs
  739. _, err = pager.NextPage(&objects)
  740. if err != nil {
  741. metric.GCSListObjectsCompleted(err)
  742. return result, err
  743. }
  744. for _, attrs := range objects {
  745. name, _ := fs.resolve(attrs.Name, prefix, attrs.ContentType)
  746. // a dir object with a trailing slash will result in an empty name
  747. if name == "/" || name == "" {
  748. continue
  749. }
  750. result = true
  751. break
  752. }
  753. metric.GCSListObjectsCompleted(nil)
  754. return result, nil
  755. }
  756. func (fs *GCSFs) getPrefix(name string) string {
  757. prefix := ""
  758. if name != "" && name != "." && name != "/" {
  759. prefix = strings.TrimPrefix(name, "/")
  760. if !strings.HasSuffix(prefix, "/") {
  761. prefix += "/"
  762. }
  763. }
  764. return prefix
  765. }
  766. func (fs *GCSFs) headObject(name string) (*storage.ObjectAttrs, error) {
  767. ctx, cancelFn := context.WithDeadline(context.Background(), time.Now().Add(fs.ctxTimeout))
  768. defer cancelFn()
  769. bkt := fs.svc.Bucket(fs.config.Bucket)
  770. obj := bkt.Object(name)
  771. attrs, err := obj.Attrs(ctx)
  772. metric.GCSHeadObjectCompleted(err)
  773. return attrs, err
  774. }
  775. // GetMimeType returns the content type
  776. func (fs *GCSFs) GetMimeType(name string) (string, error) {
  777. attrs, err := fs.headObject(name)
  778. if err != nil {
  779. return "", err
  780. }
  781. return attrs.ContentType, nil
  782. }
  783. // Close closes the fs
  784. func (fs *GCSFs) Close() error {
  785. return nil
  786. }
  787. // GetAvailableDiskSize returns the available size for the specified path
  788. func (*GCSFs) GetAvailableDiskSize(dirName string) (*sftp.StatVFS, error) {
  789. return nil, ErrStorageSizeUnavailable
  790. }
  791. func (fs *GCSFs) getStorageID() string {
  792. return fmt.Sprintf("gs://%v", fs.config.Bucket)
  793. }