aboutsummaryrefslogtreecommitdiff
path: root/weed/server/webdav_server.go
diff options
context:
space:
mode:
Diffstat (limited to 'weed/server/webdav_server.go')
-rw-r--r--weed/server/webdav_server.go141
1 files changed, 68 insertions, 73 deletions
diff --git a/weed/server/webdav_server.go b/weed/server/webdav_server.go
index d75869f30..37c4afd5c 100644
--- a/weed/server/webdav_server.go
+++ b/weed/server/webdav_server.go
@@ -1,21 +1,24 @@
package weed_server
import (
- "bytes"
"context"
"fmt"
"io"
+ "math"
"os"
"path"
"strings"
"time"
+ "github.com/chrislusf/seaweedfs/weed/util/grace"
"golang.org/x/net/webdav"
"google.golang.org/grpc"
"github.com/chrislusf/seaweedfs/weed/operation"
+ "github.com/chrislusf/seaweedfs/weed/pb"
"github.com/chrislusf/seaweedfs/weed/pb/filer_pb"
"github.com/chrislusf/seaweedfs/weed/util"
+ "github.com/chrislusf/seaweedfs/weed/util/chunk_cache"
"github.com/chrislusf/seaweedfs/weed/filer2"
"github.com/chrislusf/seaweedfs/weed/glog"
@@ -31,6 +34,9 @@ type WebDavOption struct {
Collection string
Uid uint32
Gid uint32
+ Cipher bool
+ CacheDir string
+ CacheSizeMB int64
}
type WebDavServer struct {
@@ -64,6 +70,7 @@ type WebDavFileSystem struct {
secret security.SigningKey
filer *filer2.Filer
grpcDialOption grpc.DialOption
+ chunkCache *chunk_cache.ChunkCache
}
type FileInfo struct {
@@ -88,22 +95,34 @@ type WebDavFile struct {
off int64
entry *filer_pb.Entry
entryViewCache []filer2.VisibleInterval
+ reader io.ReaderAt
}
func NewWebDavFileSystem(option *WebDavOption) (webdav.FileSystem, error) {
+
+ chunkCache := chunk_cache.NewChunkCache(256, option.CacheDir, option.CacheSizeMB)
+ grace.OnInterrupt(func() {
+ chunkCache.Shutdown()
+ })
return &WebDavFileSystem{
- option: option,
+ option: option,
+ chunkCache: chunkCache,
}, nil
}
-func (fs *WebDavFileSystem) WithFilerClient(ctx context.Context, fn func(context.Context, filer_pb.SeaweedFilerClient) error) error {
+var _ = filer_pb.FilerClient(&WebDavFileSystem{})
- return util.WithCachedGrpcClient(ctx, func(ctx2 context.Context, grpcConnection *grpc.ClientConn) error {
+func (fs *WebDavFileSystem) WithFilerClient(fn func(filer_pb.SeaweedFilerClient) error) error {
+
+ return pb.WithCachedGrpcClient(func(grpcConnection *grpc.ClientConn) error {
client := filer_pb.NewSeaweedFilerClient(grpcConnection)
- return fn(ctx2, client)
+ return fn(client)
}, fs.option.FilerGrpcAddress, fs.option.GrpcDialOption)
}
+func (fs *WebDavFileSystem) AdjustedUrl(hostAndPort string) string {
+ return hostAndPort
+}
func clearName(name string) (string, error) {
slashed := strings.HasSuffix(name, "/")
@@ -135,8 +154,8 @@ func (fs *WebDavFileSystem) Mkdir(ctx context.Context, fullDirPath string, perm
return os.ErrExist
}
- return fs.WithFilerClient(ctx, func(ctx context.Context, client filer_pb.SeaweedFilerClient) error {
- dir, name := filer2.FullPath(fullDirPath).DirAndName()
+ return fs.WithFilerClient(func(client filer_pb.SeaweedFilerClient) error {
+ dir, name := util.FullPath(fullDirPath).DirAndName()
request := &filer_pb.CreateEntryRequest{
Directory: dir,
Entry: &filer_pb.Entry{
@@ -153,7 +172,7 @@ func (fs *WebDavFileSystem) Mkdir(ctx context.Context, fullDirPath string, perm
}
glog.V(1).Infof("mkdir: %v", request)
- if err := filer_pb.CreateEntry(ctx, client, request); err != nil {
+ if err := filer_pb.CreateEntry(client, request); err != nil {
return fmt.Errorf("mkdir %s/%s: %v", dir, name, err)
}
@@ -183,9 +202,9 @@ func (fs *WebDavFileSystem) OpenFile(ctx context.Context, fullFilePath string, f
fs.removeAll(ctx, fullFilePath)
}
- dir, name := filer2.FullPath(fullFilePath).DirAndName()
- err = fs.WithFilerClient(ctx, func(ctx context.Context, client filer_pb.SeaweedFilerClient) error {
- if err := filer_pb.CreateEntry(ctx, client, &filer_pb.CreateEntryRequest{
+ dir, name := util.FullPath(fullFilePath).DirAndName()
+ err = fs.WithFilerClient(func(client filer_pb.SeaweedFilerClient) error {
+ if err := filer_pb.CreateEntry(client, &filer_pb.CreateEntryRequest{
Directory: dir,
Entry: &filer_pb.Entry{
Name: name,
@@ -238,34 +257,10 @@ func (fs *WebDavFileSystem) removeAll(ctx context.Context, fullFilePath string)
return err
}
- fi, err := fs.stat(ctx, fullFilePath)
- if err != nil {
- return err
- }
-
- if fi.IsDir() {
- //_, err = fs.db.Exec(`delete from filesystem where fullFilePath like $1 escape '\'`, strings.Replace(fullFilePath, `%`, `\%`, -1)+`%`)
- } else {
- //_, err = fs.db.Exec(`delete from filesystem where fullFilePath = ?`, fullFilePath)
- }
- dir, name := filer2.FullPath(fullFilePath).DirAndName()
- err = fs.WithFilerClient(ctx, func(ctx context.Context, client filer_pb.SeaweedFilerClient) error {
-
- request := &filer_pb.DeleteEntryRequest{
- Directory: dir,
- Name: name,
- IsDeleteData: true,
- }
+ dir, name := util.FullPath(fullFilePath).DirAndName()
- glog.V(3).Infof("removing entry: %v", request)
- _, err := client.DeleteEntry(ctx, request)
- if err != nil {
- return fmt.Errorf("remove %s: %v", fullFilePath, err)
- }
+ return filer_pb.Remove(fs, dir, name, true, false, false)
- return nil
- })
- return err
}
func (fs *WebDavFileSystem) RemoveAll(ctx context.Context, name string) error {
@@ -305,10 +300,10 @@ func (fs *WebDavFileSystem) Rename(ctx context.Context, oldName, newName string)
return os.ErrExist
}
- oldDir, oldBaseName := filer2.FullPath(oldName).DirAndName()
- newDir, newBaseName := filer2.FullPath(newName).DirAndName()
+ oldDir, oldBaseName := util.FullPath(oldName).DirAndName()
+ newDir, newBaseName := util.FullPath(newName).DirAndName()
- return fs.WithFilerClient(ctx, func(ctx context.Context, client filer_pb.SeaweedFilerClient) error {
+ return fs.WithFilerClient(func(client filer_pb.SeaweedFilerClient) error {
request := &filer_pb.AtomicRenameEntryRequest{
OldDirectory: oldDir,
@@ -333,10 +328,10 @@ func (fs *WebDavFileSystem) stat(ctx context.Context, fullFilePath string) (os.F
return nil, err
}
- fullpath := filer2.FullPath(fullFilePath)
+ fullpath := util.FullPath(fullFilePath)
var fi FileInfo
- entry, err := filer2.GetEntry(ctx, fs, fullpath)
+ entry, err := filer_pb.GetEntry(fs, fullpath)
if entry == nil {
return nil, os.ErrNotExist
}
@@ -367,10 +362,12 @@ func (f *WebDavFile) Write(buf []byte) (int, error) {
glog.V(2).Infof("WebDavFileSystem.Write %v", f.name)
+ dir, _ := util.FullPath(f.name).DirAndName()
+
var err error
ctx := context.Background()
if f.entry == nil {
- f.entry, err = filer2.GetEntry(ctx, f.fs, filer2.FullPath(f.name))
+ f.entry, err = filer_pb.GetEntry(f.fs, util.FullPath(f.name))
}
if f.entry == nil {
@@ -382,13 +379,15 @@ func (f *WebDavFile) Write(buf []byte) (int, error) {
var fileId, host string
var auth security.EncodedJwt
+ var collection, replication string
- if err = f.fs.WithFilerClient(ctx, func(ctx context.Context, client filer_pb.SeaweedFilerClient) error {
+ if err = f.fs.WithFilerClient(func(client filer_pb.SeaweedFilerClient) error {
request := &filer_pb.AssignVolumeRequest{
Count: 1,
- Replication: "000",
+ Replication: "",
Collection: f.fs.option.Collection,
+ ParentPath: dir,
}
resp, err := client.AssignVolume(ctx, request)
@@ -396,8 +395,12 @@ func (f *WebDavFile) Write(buf []byte) (int, error) {
glog.V(0).Infof("assign volume failure %v: %v", request, err)
return err
}
+ if resp.Error != "" {
+ return fmt.Errorf("assign volume failure %v: %v", request, resp.Error)
+ }
fileId, host, auth = resp.FileId, resp.Url, security.EncodedJwt(resp.Auth)
+ collection, replication = resp.Collection, resp.Replication
return nil
}); err != nil {
@@ -405,8 +408,7 @@ func (f *WebDavFile) Write(buf []byte) (int, error) {
}
fileUrl := fmt.Sprintf("http://%s/%s", host, fileId)
- bufReader := bytes.NewReader(buf)
- uploadResult, err := operation.Upload(fileUrl, f.name, bufReader, false, "", nil, auth)
+ uploadResult, err := operation.UploadData(fileUrl, f.name, f.fs.option.Cipher, buf, false, "", nil, auth)
if err != nil {
glog.V(0).Infof("upload data %v to %s: %v", f.name, fileUrl, err)
return 0, fmt.Errorf("upload data: %v", err)
@@ -416,19 +418,12 @@ func (f *WebDavFile) Write(buf []byte) (int, error) {
return 0, fmt.Errorf("upload result: %v", uploadResult.Error)
}
- chunk := &filer_pb.FileChunk{
- FileId: fileId,
- Offset: f.off,
- Size: uint64(len(buf)),
- Mtime: time.Now().UnixNano(),
- ETag: uploadResult.ETag,
- }
-
- f.entry.Chunks = append(f.entry.Chunks, chunk)
- dir, _ := filer2.FullPath(f.name).DirAndName()
+ f.entry.Chunks = append(f.entry.Chunks, uploadResult.ToPbFileChunk(fileId, f.off))
- err = f.fs.WithFilerClient(ctx, func(ctx context.Context, client filer_pb.SeaweedFilerClient) error {
+ err = f.fs.WithFilerClient(func(client filer_pb.SeaweedFilerClient) error {
f.entry.Attributes.Mtime = time.Now().Unix()
+ f.entry.Attributes.Collection = collection
+ f.entry.Attributes.Replication = replication
request := &filer_pb.UpdateEntryRequest{
Directory: dir,
@@ -465,10 +460,9 @@ func (f *WebDavFile) Close() error {
func (f *WebDavFile) Read(p []byte) (readSize int, err error) {
glog.V(2).Infof("WebDavFileSystem.Read %v", f.name)
- ctx := context.Background()
if f.entry == nil {
- f.entry, err = filer2.GetEntry(ctx, f.fs, filer2.FullPath(f.name))
+ f.entry, err = filer_pb.GetEntry(f.fs, util.FullPath(f.name))
}
if f.entry == nil {
return 0, err
@@ -481,33 +475,33 @@ func (f *WebDavFile) Read(p []byte) (readSize int, err error) {
}
if f.entryViewCache == nil {
f.entryViewCache = filer2.NonOverlappingVisibleIntervals(f.entry.Chunks)
+ f.reader = nil
}
- chunkViews := filer2.ViewFromVisibleIntervals(f.entryViewCache, f.off, len(p))
-
- totalRead, err := filer2.ReadIntoBuffer(ctx, f.fs, filer2.FullPath(f.name), p, chunkViews, f.off)
- if err != nil {
- return 0, err
+ if f.reader == nil {
+ chunkViews := filer2.ViewFromVisibleIntervals(f.entryViewCache, 0, math.MaxInt32)
+ f.reader = filer2.NewChunkReaderAtFromClient(f.fs, chunkViews, f.fs.chunkCache)
}
- readSize = int(totalRead)
- glog.V(3).Infof("WebDavFileSystem.Read %v: [%d,%d)", f.name, f.off, f.off+totalRead)
+ readSize, err = f.reader.ReadAt(p, f.off)
- f.off += totalRead
- if readSize == 0 {
- return 0, io.EOF
+ glog.V(3).Infof("WebDavFileSystem.Read %v: [%d,%d)", f.name, f.off, f.off+int64(readSize))
+ f.off += int64(readSize)
+
+ if err != nil {
+ glog.Errorf("file read %s: %v", f.name, err)
}
return
+
}
func (f *WebDavFile) Readdir(count int) (ret []os.FileInfo, err error) {
glog.V(2).Infof("WebDavFileSystem.Readdir %v count %d", f.name, count)
- ctx := context.Background()
- dir, _ := filer2.FullPath(f.name).DirAndName()
+ dir, _ := util.FullPath(f.name).DirAndName()
- err = filer2.ReadDirAllEntries(ctx, f.fs, filer2.FullPath(dir), "", func(entry *filer_pb.Entry, isLast bool) {
+ err = filer_pb.ReadDirAllEntries(f.fs, util.FullPath(dir), "", func(entry *filer_pb.Entry, isLast bool) error {
fi := FileInfo{
size: int64(filer2.TotalSize(entry.GetChunks())),
name: entry.Name,
@@ -521,6 +515,7 @@ func (f *WebDavFile) Readdir(count int) (ret []os.FileInfo, err error) {
}
glog.V(4).Infof("entry: %v", fi.name)
ret = append(ret, &fi)
+ return nil
})
old := f.off
generated by cgit v1.2.3 (git 2.25.1) at 2025-12-26 16:32:06 +0000