afero/gcsfs/fs.go

381 lines
8.7 KiB
Go

// Copyright © 2021 Vasily Ovchinnikov <vasily@remerge.io>.
//
// The code in this file is derived from afero fork github.com/Zatte/afero by Mikael Rapp
// licensed under Apache License 2.0.
//
// Licensed under the Apache License, Version 2.0 (the "License");
// you may not use this file except in compliance with the License.
// You may obtain a copy of the License at
// http://www.apache.org/licenses/LICENSE-2.0
//
// Unless required by applicable law or agreed to in writing, software
// distributed under the License is distributed on an "AS IS" BASIS,
// WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
// See the License for the specific language governing permissions and
// limitations under the License.
package gcsfs
import (
"context"
"errors"
"os"
"path/filepath"
"strings"
"syscall"
"time"
"github.com/googleapis/google-cloud-go-testing/storage/stiface"
)
const (
defaultFileMode = 0755
gsPrefix = "gs://"
)
// GcsFs is a Fs implementation that uses functions provided by google cloud storage
type GcsFs struct {
ctx context.Context
client stiface.Client
separator string
buckets map[string]stiface.BucketHandle
rawGcsObjects map[string]*GcsFile
autoRemoveEmptyFolders bool //trigger for creating "virtual folders" (not required by GCSs)
}
func NewGcsFs(ctx context.Context, client stiface.Client) *GcsFs {
return NewGcsFsWithSeparator(ctx, client, "/")
}
func NewGcsFsWithSeparator(ctx context.Context, client stiface.Client, folderSep string) *GcsFs {
return &GcsFs{
ctx: ctx,
client: client,
separator: folderSep,
rawGcsObjects: make(map[string]*GcsFile),
autoRemoveEmptyFolders: true,
}
}
// normSeparators will normalize all "\\" and "/" to the provided separator
func (fs *GcsFs) normSeparators(s string) string {
return strings.Replace(strings.Replace(s, "\\", fs.separator, -1), "/", fs.separator, -1)
}
func (fs *GcsFs) ensureTrailingSeparator(s string) string {
if len(s) > 0 && !strings.HasSuffix(s, fs.separator) {
return s + fs.separator
}
return s
}
func (fs *GcsFs) ensureNoLeadingSeparator(s string) string {
if len(s) > 0 && strings.HasPrefix(s, fs.separator) {
s = s[len(fs.separator):]
}
return s
}
func ensureNoPrefix(s string) string {
if len(s) > 0 && strings.HasPrefix(s, gsPrefix) {
return s[len(gsPrefix):]
}
return s
}
func validateName(s string) error {
if len(s) == 0 {
return ErrNoBucketInName
}
return nil
}
// Splits provided name into bucket name and path
func (fs *GcsFs) splitName(name string) (bucketName string, path string) {
splitName := strings.Split(name, fs.separator)
return splitName[0], strings.Join(splitName[1:], fs.separator)
}
func (fs *GcsFs) getBucket(name string) (stiface.BucketHandle, error) {
bucket := fs.buckets[name]
if bucket == nil {
bucket = fs.client.Bucket(name)
_, err := bucket.Attrs(fs.ctx)
if err != nil {
return nil, err
}
}
return bucket, nil
}
func (fs *GcsFs) getObj(name string) (stiface.ObjectHandle, error) {
bucketName, path := fs.splitName(name)
bucket, err := fs.getBucket(bucketName)
if err != nil {
return nil, err
}
return bucket.Object(path), nil
}
func (fs *GcsFs) Name() string { return "GcsFs" }
func (fs *GcsFs) Create(name string) (*GcsFile, error) {
name = fs.ensureNoLeadingSeparator(fs.normSeparators(ensureNoPrefix(name)))
if err := validateName(name); err != nil {
return nil, err
}
if !fs.autoRemoveEmptyFolders {
baseDir := filepath.Base(name)
if stat, err := fs.Stat(baseDir); err != nil || !stat.IsDir() {
err = fs.MkdirAll(baseDir, 0)
if err != nil {
return nil, err
}
}
}
obj, err := fs.getObj(name)
if err != nil {
return nil, err
}
w := obj.NewWriter(fs.ctx)
err = w.Close()
if err != nil {
return nil, err
}
file := NewGcsFile(fs.ctx, fs, obj, os.O_RDWR|os.O_CREATE|os.O_TRUNC, 0, name)
fs.rawGcsObjects[name] = file
return file, nil
}
func (fs *GcsFs) Mkdir(name string, _ os.FileMode) error {
name = fs.ensureNoLeadingSeparator(fs.ensureTrailingSeparator(fs.normSeparators(ensureNoPrefix(name))))
if err := validateName(name); err != nil {
return err
}
obj, err := fs.getObj(name)
if err != nil {
return err
}
w := obj.NewWriter(fs.ctx)
return w.Close()
}
func (fs *GcsFs) MkdirAll(path string, perm os.FileMode) error {
path = fs.ensureNoLeadingSeparator(fs.ensureTrailingSeparator(fs.normSeparators(ensureNoPrefix(path))))
if err := validateName(path); err != nil {
return err
}
root := ""
folders := strings.Split(path, fs.separator)
for i, f := range folders {
if f == "" && i != 0 {
continue // it's the last item - it should be empty
}
//Don't force a delimiter prefix
if root != "" {
root = root + fs.separator + f
} else {
root = f
}
if err := fs.Mkdir(root, perm); err != nil {
return err
}
}
return nil
}
func (fs *GcsFs) Open(name string) (*GcsFile, error) {
return fs.OpenFile(name, os.O_RDONLY, 0)
}
func (fs *GcsFs) OpenFile(name string, flag int, fileMode os.FileMode) (*GcsFile, error) {
var file *GcsFile
var err error
name = fs.ensureNoLeadingSeparator(fs.normSeparators(ensureNoPrefix(name)))
if err = validateName(name); err != nil {
return nil, err
}
f, found := fs.rawGcsObjects[name]
if found {
file = NewGcsFileFromOldFH(flag, fileMode, f.resource)
} else {
var obj stiface.ObjectHandle
obj, err = fs.getObj(name)
if err != nil {
return nil, err
}
file = NewGcsFile(fs.ctx, fs, obj, flag, fileMode, name)
}
if flag == os.O_RDONLY {
_, err = file.Stat()
if err != nil {
return nil, err
}
}
if flag&os.O_TRUNC != 0 {
err = file.resource.obj.Delete(fs.ctx)
if err != nil {
return nil, err
}
return fs.Create(name)
}
if flag&os.O_APPEND != 0 {
_, err = file.Seek(0, 2)
if err != nil {
return nil, err
}
}
if flag&os.O_CREATE != 0 {
_, err = file.Stat()
if err == nil { // the file actually exists
return nil, syscall.EPERM
}
_, err = file.WriteString("")
if err != nil {
return nil, err
}
}
return file, nil
}
func (fs *GcsFs) Remove(name string) error {
name = fs.ensureNoLeadingSeparator(fs.normSeparators(ensureNoPrefix(name)))
if err := validateName(name); err != nil {
return err
}
obj, err := fs.getObj(name)
if err != nil {
return err
}
info, err := fs.Stat(name)
if err != nil {
return err
}
delete(fs.rawGcsObjects, name)
if info.IsDir() {
// it's a folder, we ha to check its contents - it cannot be removed, if not empty
var dir *GcsFile
dir, err = fs.Open(name)
if err != nil {
return err
}
var infos []os.FileInfo
infos, err = dir.Readdir(0)
if len(infos) > 0 {
return syscall.ENOTEMPTY
}
// it's an empty folder, we can continue
name = fs.ensureTrailingSeparator(name)
obj, err = fs.getObj(name)
if err != nil {
return err
}
return obj.Delete(fs.ctx)
}
return obj.Delete(fs.ctx)
}
func (fs *GcsFs) RemoveAll(path string) error {
path = fs.ensureNoLeadingSeparator(fs.normSeparators(ensureNoPrefix(path)))
if err := validateName(path); err != nil {
return err
}
pathInfo, err := fs.Stat(path)
if err != nil {
return err
}
if !pathInfo.IsDir() {
return fs.Remove(path)
}
var dir *GcsFile
dir, err = fs.Open(path)
if err != nil {
return err
}
var infos []os.FileInfo
infos, err = dir.Readdir(0)
for _, info := range infos {
nameToRemove := fs.normSeparators(info.Name())
err = fs.RemoveAll(path + fs.separator + nameToRemove)
if err != nil {
return err
}
}
return fs.Remove(path)
}
func (fs *GcsFs) Rename(oldName, newName string) error {
oldName = fs.ensureNoLeadingSeparator(fs.normSeparators(ensureNoPrefix(oldName)))
if err := validateName(oldName); err != nil {
return err
}
newName = fs.ensureNoLeadingSeparator(fs.normSeparators(ensureNoPrefix(newName)))
if err := validateName(newName); err != nil {
return err
}
src, err := fs.getObj(oldName)
if err != nil {
return err
}
dst, err := fs.getObj(newName)
if err != nil {
return err
}
if _, err = dst.CopierFrom(src).Run(fs.ctx); err != nil {
return err
}
delete(fs.rawGcsObjects, oldName)
return src.Delete(fs.ctx)
}
func (fs *GcsFs) Stat(name string) (os.FileInfo, error) {
name = fs.ensureNoLeadingSeparator(fs.normSeparators(ensureNoPrefix(name)))
if err := validateName(name); err != nil {
return nil, err
}
return newFileInfo(name, fs, defaultFileMode)
}
func (fs *GcsFs) Chmod(_ string, _ os.FileMode) error {
return errors.New("method Chmod is not implemented in GCS")
}
func (fs *GcsFs) Chtimes(_ string, _, _ time.Time) error {
return errors.New("method Chtimes is not implemented. Create, Delete, Updated times are read only fields in GCS and set implicitly")
}
func (fs *GcsFs) Chown(_ string, _, _ int) error {
return errors.New("method Chown is not implemented for GCS")
}