gcsfs.go 22 KB

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