From b5ae5504a75c4fd2252e4f67f8b6450372256c05 Mon Sep 17 00:00:00 2001 From: ston1th Date: Tue, 22 Mar 2022 21:48:59 +0100 Subject: [PATCH] added cache only access server --- cmd/cachefs/main.go | 35 +++++++++++++- pkg/fs/fs.go | 41 ++++++++++++++++- pkg/srv/cache.go | 102 +++++++++++++++++++++++++++++++++++++++++ pkg/srv/interceptor.go | 5 +- pkg/srv/srv.go | 2 +- pkg/srv/templates.go | 40 ++++++++++++++-- 6 files changed, 217 insertions(+), 8 deletions(-) create mode 100644 pkg/srv/cache.go diff --git a/cmd/cachefs/main.go b/cmd/cachefs/main.go index 8072c1b..3c5aa38 100644 --- a/cmd/cachefs/main.go +++ b/cmd/cachefs/main.go @@ -27,6 +27,7 @@ var ( metadata string listenHttp string listenWebdav string + listenCache string quota int64 log logr.Logger @@ -41,6 +42,7 @@ func main() { 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") + flag.StringVar(&listenCache, "cache", "", "listen addr:port for cache only") flag.Int64Var("a, "quota", 1, "max disk usage quota in GiB") flag.Parse() @@ -51,14 +53,23 @@ func main() { if err != nil { klog.Fatalf("init failed: %s", err) } - s := &http.Server{Addr: listenHttp, Handler: srv.NewFileServer(filesystem, klogr.New().WithName("server"))} + s := &http.Server{ + Addr: listenHttp, + Handler: srv.NewFileServer( + filesystem, + klogr.New().WithName("server"), + ), + } go func() { err := s.ListenAndServe() if err != nil && err != http.ErrServerClosed { klog.Fatalf("init failed: %s", err) } }() - var dav *http.Server + var ( + dav *http.Server + cache *http.Server + ) if listenWebdav != "" { dav = &http.Server{ Addr: listenWebdav, @@ -74,6 +85,21 @@ func main() { } }() } + if listenCache != "" { + cache = &http.Server{ + Addr: listenCache, + Handler: srv.NewCacheServer( + filesystem, + klogr.New().WithName("cache"), + ), + } + go func() { + err := cache.ListenAndServe() + if err != nil && err != http.ErrServerClosed { + klog.Fatalf("init failed: %s", err) + } + }() + } sigs := make(chan os.Signal, 1) signal.Notify(sigs, syscall.SIGINT, syscall.SIGTERM) <-sigs @@ -86,6 +112,11 @@ func main() { dav.Shutdown(ctx) cancel() } + if cache != nil { + ctx, cancel := context.WithTimeout(context.Background(), time.Second*5) + cache.Shutdown(ctx) + cancel() + } filesystem.Close() log.Info("cachefs shutdown completed") os.Exit(0) diff --git a/pkg/fs/fs.go b/pkg/fs/fs.go index d57f48c..abf0598 100644 --- a/pkg/fs/fs.go +++ b/pkg/fs/fs.go @@ -128,11 +128,50 @@ func (fs *FS) RemoveDst(name string) error { return os.Remove(dp) } -func (fs *FS) statDst(name string) (fi stdfs.FileInfo, err error) { +func (fs *FS) StatDst(name string) (fi stdfs.FileInfo, err error) { + fi, err = fs.sc.Get(name) + 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) } +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) { + log.Error(err, "error stat cache file") + } + return + } + file, err := os.Open(dp) + if err != nil { + if !skipLog(name) { + log.Error(err, "error opening cache file") + } + return + } + if fi.IsDir() { + f = &Dir{log: log.WithName("dir"), f: file, dc: fs.dc} + return + } + md := fs.metadata(name, fi.Size()) + f = &File{ + log: log, + f: file, + md: md, + offline: true, + } + return +} + func (fs *FS) CancelPreload(name string) { if cancel, ok := fs.pm[name]; ok { cancel() diff --git a/pkg/srv/cache.go b/pkg/srv/cache.go new file mode 100644 index 0000000..69d191b --- /dev/null +++ b/pkg/srv/cache.go @@ -0,0 +1,102 @@ +// Copyright (C) 2022 Marius Schellenberger + +package srv + +import ( + "io" + "net/http" + "path" + "strings" + + "cachefs/pkg/fs" + + "github.com/go-logr/logr" +) + +type CacheFS struct { + fs *fs.FS +} + +func (cfs *CacheFS) Open(name string) (http.File, error) { + return cfs.fs.OpenDst(name) +} + +type CacheServer struct { + log logr.Logger + fs *fs.FS + h http.Handler +} + +func NewCacheServer(fs *fs.FS, log logr.Logger) http.Handler { + return &CacheServer{log, fs, http.FileServer(&CacheFS{fs})} +} + +func (cs *CacheServer) ServeHTTP(w http.ResponseWriter, r *http.Request) { + s := &statusInterceptor{w: w} + w = s + if !skipLog(r.RequestURI) { + defer func() { + if s.Status() == http.StatusPartialContent { + return + } + cs.log.Info("access", + "client", r.RemoteAddr, + "method", r.Method, + "status", s.Status(), + "uri", r.RequestURI, + ) + }() + } + upath := r.URL.Path + if !strings.HasPrefix(upath, "/") { + upath = "/" + upath + r.URL.Path = upath + } + p := path.Clean(upath) + option := r.FormValue("o") + d, err := cs.fs.StatDst(p) + if err != nil { + msg, code := toHTTPError(err) + http.Error(w, msg, code) + return + } + h := w.Header() + if option == "v" { + h.Set(csp, videoCSP) + err = video.Execute(w, r.URL.Path) + if err != nil { + http.Error(w, err.Error(), http.StatusInternalServerError) + } + return + } + + if !d.IsDir() { + cs.h.ServeHTTP(w, r) + return + } + + i := &responseInterceptor{w: w} + r.Header.Del("If-Modified-Since") + r.Header.Del("Cache-Control") + cs.h.ServeHTTP(i, r) + + h.Del("Last-Modified") + + paths, err := i.GetPaths(p, cs.fs, true) + if err == io.EOF { + w.WriteHeader(i.Status()) + return + } + if err != nil { + http.Error(w, err.Error(), http.StatusInternalServerError) + return + } + h.Set(csp, indexCSP) + + w.WriteHeader(i.Status()) + err = cache.Execute(w, paths) + if err != nil { + http.Error(w, err.Error(), http.StatusInternalServerError) + return + } +} diff --git a/pkg/srv/interceptor.go b/pkg/srv/interceptor.go index 1b717fc..31ab393 100644 --- a/pkg/srv/interceptor.go +++ b/pkg/srv/interceptor.go @@ -91,7 +91,7 @@ func (r *responseInterceptor) Status() int { return r.status } -func (r *responseInterceptor) GetPaths(path string, fs *fs.FS) (dc dirContents, err error) { +func (r *responseInterceptor) GetPaths(path string, fs *fs.FS, relative bool) (dc dirContents, err error) { buf := r.buf.Bytes() buf = bytes.ReplaceAll(buf, []byte{'&'}, []byte("&")) err = xml.Unmarshal(buf, &dc) @@ -106,6 +106,9 @@ func (r *responseInterceptor) GetPaths(path string, fs *fs.FS) (dc dirContents, name := template.HTML(p) full := path + "/" + p uri := template.HTML(full) + if relative { + uri = template.HTML(p) + } if p[len(p)-1] == '/' { dc.Dirs = append(dc.Dirs, dir{ Name: name, diff --git a/pkg/srv/srv.go b/pkg/srv/srv.go index dc5c7d7..66fed6a 100644 --- a/pkg/srv/srv.go +++ b/pkg/srv/srv.go @@ -103,7 +103,7 @@ func (fs *FileServer) ServeHTTP(w http.ResponseWriter, r *http.Request) { h.Del("Last-Modified") - paths, err := i.GetPaths(p, fs.fs) + paths, err := i.GetPaths(p, fs.fs, false) if err == io.EOF { w.WriteHeader(i.Status()) return diff --git a/pkg/srv/templates.go b/pkg/srv/templates.go index c05d348..6d344c7 100644 --- a/pkg/srv/templates.go +++ b/pkg/srv/templates.go @@ -5,7 +5,7 @@ package srv import "html/template" var ( - index = template.Must(template.New("index").Parse(` + indexHead = ` @@ -107,8 +107,9 @@ span { font-size: 15px; } - - +` + + index = template.Must(template.New("index").Parse(indexHead + ` 
[v]: show video
 [n]: skip file caching
@@ -143,6 +144,39 @@ span {
+`)) + cache = template.Must(template.New("cache").Parse(indexHead + ` +
+ 
[v]: show video
+ + + + + + +{{range $s := .Dirs -}} + + + + + +{{end -}} +{{range $s := .Files -}} + + + + + +{{end -}} +
../[dir]
{{$s.Name}}[dir]
{{$s.Name}} + {{if ge $s.Status 0}}{{$s.Status}}%{{end}}[v] +
+{{if ge $s.Status 0}} +
+{{end -}} +
+
+ `)) video = template.Must(template.New("video").Parse(`