diff --git a/cmd/cachefs/main.go b/cmd/cachefs/main.go index aad2c37..7a28652 100644 --- a/cmd/cachefs/main.go +++ b/cmd/cachefs/main.go @@ -39,8 +39,8 @@ const gib = 1024 * 1024 * 1024 func main() { klog.InitFlags(nil) - flag.StringVar(&src, "src", "", "root directory (NFS share)") - flag.StringVar(&dst, "dst", "", "local directory to cache files") + flag.StringVar(&src, "src", "", "url path to source files (example: file:///mnt/nfs)") + flag.StringVar(&dst, "dst", "", "url path to cache files (example: file:///mnt/cache)") flag.StringVar(&metadata, "data", "", "path to metadata file") flag.StringVar(&listenHttp, "listen", "127.0.0.1:8080", "listen addr:port") flag.StringVar(&listenWebdav, "webdav", "", "listen addr:port for webdav") diff --git a/cmd/metadatagen/main.go b/cmd/metadatagen/main.go index 39d5efd..28f7898 100644 --- a/cmd/metadatagen/main.go +++ b/cmd/metadatagen/main.go @@ -8,6 +8,7 @@ import ( "os" "cachefs/pkg/fs" + "cachefs/pkg/provider/parse" ) var ( @@ -18,11 +19,16 @@ var ( ) func main() { - flag.StringVar(&dst, "dst", "", "local directory to cache files") + flag.StringVar(&dst, "dst", "", "url path to cache files (example: file:///mnt/cache)") flag.StringVar(&metadata, "data", "", "path to metadata file") flag.Parse() - err := fs.MetadataGenerator(metadata, dst) + dstfs, err := parse.FS(dst) + if err != nil { + fmt.Println(err) + os.Exit(1) + } + err = fs.MetadataGenerator(metadata, dstfs) if err != nil { fmt.Println(err) os.Exit(1) diff --git a/pkg/fs/dir.go b/pkg/fs/dir.go index 113b7fa..24d8fad 100644 --- a/pkg/fs/dir.go +++ b/pkg/fs/dir.go @@ -3,6 +3,7 @@ package fs import ( + "cachefs/pkg/provider" "errors" "io" "os" @@ -45,7 +46,7 @@ func (dc *DirCache) Set(name string, dce *DirCacheEntry) { type Dir struct { log logr.Logger - f *os.File + f provider.File dc *DirCache } @@ -57,7 +58,7 @@ func (f *Dir) Seek(_ int64, _ int) (int64, error) { return 0, io.EOF } -func readDir(f *os.File) (fi []os.FileInfo, err error) { +func readDir(f provider.File) (fi []os.FileInfo, err error) { ent, err := f.ReadDir(0) if err != nil { return diff --git a/pkg/fs/file.go b/pkg/fs/file.go index 33400d9..5538b25 100644 --- a/pkg/fs/file.go +++ b/pkg/fs/file.go @@ -3,6 +3,7 @@ package fs import ( + "cachefs/pkg/provider" "context" "errors" "io" @@ -13,7 +14,7 @@ import ( type File struct { log logr.Logger - f *os.File + f provider.File md *Metadata offset int64 offline bool diff --git a/pkg/fs/fs.go b/pkg/fs/fs.go index f590784..398cac5 100644 --- a/pkg/fs/fs.go +++ b/pkg/fs/fs.go @@ -3,14 +3,13 @@ package fs import ( + "cachefs/pkg/provider" + "cachefs/pkg/provider/parse" "context" "errors" - "io" stdfs "io/fs" "net/http" "os" - "path" - "path/filepath" "strings" "github.com/go-logr/logr" @@ -21,8 +20,8 @@ type FS struct { done chan struct{} cancel func() NoCache http.Handler - src string - dst string + src provider.FS + dst provider.FS dc *DirCache sc *StatCache @@ -32,15 +31,17 @@ type FS struct { } func NewFS(quota int64, max int, src, dst, metadata string, log logr.Logger) (fs *FS, err error) { - if !filepath.IsAbs(src) { - return nil, errors.New("src path is not absolute") - } - if !filepath.IsAbs(dst) { - return nil, errors.New("dst path is not absolute") - } if src == dst { return nil, errors.New("src and dst path can not be equal") } + srcfs, err := parse.FS(src) + if err != nil { + return + } + dstfs, err := parse.FS(dst) + if err != nil { + return + } ctx, cancel := context.WithCancel(context.Background()) @@ -49,8 +50,8 @@ func NewFS(quota int64, max int, src, dst, metadata string, log logr.Logger) (fs done: make(chan struct{}), cancel: cancel, NoCache: http.FileServer(http.Dir(src)), - src: src, - dst: dst, + src: srcfs, + dst: dstfs, dc: NewDirCache(), } @@ -68,23 +69,21 @@ func NewFS(quota int64, max int, src, dst, metadata string, log logr.Logger) (fs } func (fs *FS) Stat(name string) (fi stdfs.FileInfo, err error) { - sp, dp := fs.paths(name) fi, err = fs.sc.Get(name) if err == nil { return } - fi, err = os.Stat(sp) + fi, err = fs.src.Stat(name) if err == nil { fs.sc.Set(name, fi) return } - fi, err = os.Stat(dp) + fi, err = fs.dst.Stat(name) return } func (fs *FS) RemoveDst(name string) error { - _, dp := fs.paths(name) - return os.Remove(dp) + return fs.dst.Remove(name) } func (fs *FS) StatDst(name string) (fi stdfs.FileInfo, err error) { @@ -92,17 +91,11 @@ func (fs *FS) StatDst(name string) (fi stdfs.FileInfo, err error) { if err == nil { return } - return fs.statDst(name) -} - -func (fs *FS) statDst(name string) (stdfs.FileInfo, error) { - _, dp := fs.paths(name) - return os.Stat(dp) + return fs.dst.Stat(name) } func (fs *FS) OpenDst(name string) (f http.File, err error) { log := fs.log.WithValues("file", name) - _, dp := fs.paths(name) fi, err := fs.StatDst(name) if err != nil { if !skipLog(name) { @@ -110,7 +103,7 @@ func (fs *FS) OpenDst(name string) (f http.File, err error) { } return } - file, err := os.Open(dp) + file, err := fs.dst.Open(name) if err != nil { if !skipLog(name) { log.Error(err, "error opening cache file") @@ -151,15 +144,8 @@ func skipLog(name string) bool { return strings.HasSuffix(name, "index.html") || strings.HasSuffix(name, "favicon.ico") } -func (fs *FS) paths(name string) (sp string, dp string) { - p := filepath.FromSlash(path.Clean("/" + name)) - sp = filepath.Join(fs.src, p) - dp = filepath.Join(fs.dst, p) - return -} - -func (fs *FS) open(name, sp, dp string) (f *os.File, fi os.FileInfo, offline bool, err error) { - f, err = os.Open(sp) +func (fs *FS) open(name string) (f provider.File, fi os.FileInfo, offline bool, err error) { + f, err = fs.src.Open(name) if err == nil { fi, err = f.Stat() if err == nil { @@ -167,7 +153,7 @@ func (fs *FS) open(name, sp, dp string) (f *os.File, fi os.FileInfo, offline boo } f.Close() } - f, err = os.Open(dp) + f, err = fs.dst.Open(name) if err != nil { return } @@ -181,8 +167,7 @@ func (fs *FS) open(name, sp, dp string) (f *os.File, fi os.FileInfo, offline boo func (fs *FS) Open(name string) (f http.File, err error) { log := fs.log.WithValues("file", name) - sp, dp := fs.paths(name) - file, fi, offline, err := fs.open(name, sp, dp) + file, fi, offline, err := fs.open(name) if err != nil { if !skipLog(name) { log.Error(err, "error opening source file") @@ -191,13 +176,13 @@ func (fs *FS) Open(name string) (f http.File, err error) { } if fi.IsDir() { if offline { - log.V(2).Info("dir offline mode", "path", dp) + log.V(2).Info("dir offline mode") } f = &Dir{log: log.WithName("dir"), f: file, dc: fs.dc} return } if offline { - log.V(2).Info("file offline mode", "path", dp) + log.V(2).Info("file offline mode") } md := fs.mh.Metadata(name, fi.Size()) f = &File{ @@ -209,23 +194,22 @@ func (fs *FS) Open(name string) (f http.File, err error) { return } -func (fs *FS) openCacheFile(name string, size int64) (df *os.File, err error) { - _, dp := fs.paths(name) +func (fs *FS) openCacheFile(name string, size int64) (df provider.File, err error) { log := fs.log - if i := strings.LastIndex(dp, "/"); i > 0 { - dir := dp[:i] - _, err = os.Stat(dir) + if i := strings.LastIndex(name, "/"); i > 0 { + dir := name[:i] + _, err = fs.dst.Stat(dir) if errors.Is(err, os.ErrNotExist) { - err = os.MkdirAll(dir, 0o755) + err = fs.dst.MkdirAll(dir, 0o755) if err != nil { - log.Error(err, "error creating cache dir", "dir", dp) + log.Error(err, "error creating cache dir") return } } } - _, err = os.Stat(dp) + _, err = fs.dst.Stat(name) truncate := errors.Is(err, os.ErrNotExist) - df, err = os.OpenFile(dp, os.O_RDWR|os.O_CREATE, 0o644) + df, err = fs.dst.OpenFile(name, os.O_RDWR|os.O_CREATE, 0o644) if err != nil { log.Error(err, "error opening cache file") return @@ -253,17 +237,3 @@ func (fs *FS) Close() { fs.cancel() <-fs.done } - -func dirEmpty(name string) (bool, error) { - f, err := os.Open(name) - if err != nil { - return false, err - } - _, err = f.ReadDir(1) - if err == io.EOF { - f.Close() - return true, nil - } - f.Close() - return false, err -} diff --git a/pkg/fs/metadata.go b/pkg/fs/metadata.go index 97a0ad8..1d6062b 100644 --- a/pkg/fs/metadata.go +++ b/pkg/fs/metadata.go @@ -3,10 +3,12 @@ package fs import ( + "cachefs/pkg/provider" "context" "encoding/json" "errors" "io" + "io/fs" stdfs "io/fs" "math" "os" @@ -18,7 +20,6 @@ import ( "github.com/go-logr/logr" "golang.org/x/exp/maps" - "golang.org/x/sys/unix" ) func now() int64 { @@ -35,13 +36,10 @@ type MetadataHandler struct { enc *json.Encoder } -func MetadataGenerator(file, dst string) error { +func MetadataGenerator(file string, dst provider.FS) error { if !filepath.IsAbs(file) { return errors.New("metadata path is not absolute") } - if !filepath.IsAbs(dst) { - return errors.New("dst path is not absolute") - } f, err := os.OpenFile(file, os.O_RDWR|os.O_CREATE, 0o640) if err != nil { return err @@ -53,16 +51,15 @@ func MetadataGenerator(file, dst string) error { } enc := json.NewEncoder(f) md := make(map[string]*Metadata) - filepath.WalkDir(dst, func(path string, d stdfs.DirEntry, err error) error { + fs.WalkDir(provider.WalkFS(dst), ".", func(path string, d stdfs.DirEntry, err error) error { if err != nil { return nil } if d.IsDir() { return nil } - var stat unix.Stat_t - f, err := os.Open(path) + f, err := dst.Open(path) if err != nil { return nil } @@ -72,12 +69,12 @@ func MetadataGenerator(file, dst string) error { return nil } size := fi.Size() - err = unix.Fstat(int(f.Fd()), &stat) + blocks, err := dst.Fstat(f.Fd()) if err != nil { return nil } - if size <= stat.Blocks*512 { - md[strings.TrimPrefix(path, dst)] = &Metadata{ + if size <= blocks*512 { + md[strings.TrimPrefix(path, dst.Root())] = &Metadata{ Size: size, Chunks: Chunks{{0, size}}, } @@ -188,7 +185,7 @@ func (mh *MetadataHandler) Metadata(name string, size int64) (md *Metadata) { func (mh *MetadataHandler) init() { log := mh.log maps.DeleteFunc(mh.md, func(k string, v *Metadata) bool { - _, err := mh.fs.statDst(k) + _, err := mh.fs.dst.Stat(k) if errors.Is(err, os.ErrNotExist) { log.V(2).Info("removing not existing file from metadata", "file", k) return true @@ -222,18 +219,18 @@ func (mh *MetadataHandler) flusher(ctx context.Context) { } func (mh *MetadataHandler) cleanupEmptyDirs() { - filepath.WalkDir(mh.dst, func(path string, d stdfs.DirEntry, err error) error { + fs.WalkDir(provider.WalkFS(mh.fs.dst), ".", func(path string, d stdfs.DirEntry, err error) error { if err == nil && d.IsDir() { if path == mh.dst { return nil } - empty, err := dirEmpty(path) + empty, err := mh.dirEmpty(path) if err != nil || !empty { return nil } log := mh.log.WithValues("dir", path) log.V(2).Info("removing empty dir") - err = os.Remove(path) + err = mh.fs.dst.Remove(path) if err != nil { log.Error(err, "failed to remove empty dir") } @@ -242,6 +239,19 @@ func (mh *MetadataHandler) cleanupEmptyDirs() { }) } +func (mh *MetadataHandler) dirEmpty(name string) (bool, error) { + f, err := mh.fs.dst.Open(name) + if err != nil { + return false, err + } + defer f.Close() + _, err = f.ReadDir(1) + if err == io.EOF { + return true, nil + } + return false, err +} + func (mh *MetadataHandler) flush() { log := mh.log mh.mu.Lock() @@ -306,13 +316,13 @@ func IsIOErr(err error) bool { } type Metadata struct { - mu sync.RWMutex `json:"-"` - fs *FS `json:"-"` - f *os.File `json:"-"` - name string `json:"-"` - Size int64 `json:"s"` - Atime int64 `json:"a"` - Chunks Chunks `json:"c"` + mu sync.RWMutex `json:"-"` + fs *FS `json:"-"` + f provider.File `json:"-"` + name string `json:"-"` + Size int64 `json:"s"` + Atime int64 `json:"a"` + Chunks Chunks `json:"c"` } func (md *Metadata) Delete() error { diff --git a/pkg/fs/preload.go b/pkg/fs/preload.go index 664f854..92619bd 100644 --- a/pkg/fs/preload.go +++ b/pkg/fs/preload.go @@ -130,6 +130,8 @@ func (ph *PreloadHandler) preload(ctx context.Context) { p.cancel = cancel ph.mu.Unlock() go f.Preload(ctx, func() { + file := f + file.Close() ph.fin <- name }) } diff --git a/pkg/provider/interface.go b/pkg/provider/interface.go new file mode 100644 index 0000000..6327471 --- /dev/null +++ b/pkg/provider/interface.go @@ -0,0 +1,45 @@ +// Copyright (C) 2022 Marius Schellenberger + +package provider + +import ( + "io/fs" + "os" +) + +type FS interface { + Stat(string) (fs.FileInfo, error) + Remove(string) error + Open(string) (File, error) + OpenFile(string, int, os.FileMode) (File, error) + MkdirAll(string, os.FileMode) error + Root() string + Fstat(uintptr) (int64, error) +} + +type walk struct { + fs FS +} + +func (w *walk) Open(p string) (fs.File, error) { + return w.fs.Open(p) +} + +func WalkFS(fs FS) fs.FS { + return &walk{fs} +} + +type File interface { + Name() string + Stat() (fs.FileInfo, error) + Truncate(int64) error + Sync() error + Fd() uintptr + ReadDir(int) ([]fs.DirEntry, error) + Read([]byte) (int, error) + ReadAt([]byte, int64) (int, error) + WriteAt([]byte, int64) (int, error) + Seek(int64, int) (int64, error) + Readdir(int) ([]fs.FileInfo, error) + Close() error +} diff --git a/pkg/provider/mount/mount.go b/pkg/provider/mount/mount.go new file mode 100644 index 0000000..7c2e1a6 --- /dev/null +++ b/pkg/provider/mount/mount.go @@ -0,0 +1,57 @@ +// Copyright (C) 2022 Marius Schellenberger + +package mount + +import ( + "cachefs/pkg/provider" + "io/fs" + "os" + "path" + "path/filepath" + + "golang.org/x/sys/unix" +) + +var _ provider.FS = (*FS)(nil) + +type FS struct { + root string +} + +func NewFS(root string) *FS { + return &FS{root} +} + +func (fs *FS) path(p string) string { + return filepath.Join(fs.root, filepath.FromSlash(path.Clean("/"+p))) +} + +func (fs *FS) Stat(p string) (fs.FileInfo, error) { + return os.Stat(fs.path(p)) +} + +func (fs *FS) Remove(p string) error { + return os.Remove(fs.path(p)) +} + +func (fs *FS) Open(p string) (provider.File, error) { + return os.Open(fs.path(p)) +} + +func (fs *FS) OpenFile(p string, flags int, mode os.FileMode) (provider.File, error) { + return os.OpenFile(fs.path(p), flags, mode) +} + +func (fs *FS) MkdirAll(p string, mode os.FileMode) error { + return os.MkdirAll(fs.path(p), mode) +} + +func (fs *FS) Root() string { + return fs.root +} + +func (fs *FS) Fstat(fd uintptr) (int64, error) { + var stat unix.Stat_t + err := unix.Fstat(int(fd), &stat) + return stat.Blocks, err +} diff --git a/pkg/provider/parse/parse.go b/pkg/provider/parse/parse.go new file mode 100644 index 0000000..bc00b96 --- /dev/null +++ b/pkg/provider/parse/parse.go @@ -0,0 +1,34 @@ +// Copyright (C) 2022 Marius Schellenberger + +package parse + +import ( + "cachefs/pkg/provider" + "cachefs/pkg/provider/mount" + "errors" + "fmt" + neturl "net/url" + "path/filepath" +) + +var ( + ErrPathNotAbsolute = errors.New("path is not absolute") + ErrUnsupportedScheme = errors.New("unsupported scheme") +) + +func FS(url string) (provider.FS, error) { + u, err := neturl.Parse(url) + if err != nil { + return nil, err + } + if !filepath.IsAbs(u.Path) { + return nil, ErrPathNotAbsolute + } + switch u.Scheme { + case "file": + return mount.NewFS(u.Path), nil + //case "sftp": + // return sftp.FS(u.Path), nil + } + return nil, fmt.Errorf("%w: %s", ErrUnsupportedScheme, u.Scheme) +} diff --git a/pkg/provider/sftp/sftp.go b/pkg/provider/sftp/sftp.go new file mode 100644 index 0000000..258e4fc --- /dev/null +++ b/pkg/provider/sftp/sftp.go @@ -0,0 +1,5 @@ +// Copyright (C) 2022 Marius Schellenberger + +package sftp + +// github.com/pkg/sftp