gcsfs.go 23 KB

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