gcsfs.go 22 KB

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