You can not select more than 25 topics Topics must start with a letter or number, can include dashes ('-') and can be up to 35 characters long.

569 lines
14 KiB

5 years ago
5 years ago
5 years ago
5 years ago
5 years ago
5 years ago
5 years ago
5 years ago
5 years ago
5 years ago
5 years ago
5 years ago
5 years ago
5 years ago
5 years ago
5 years ago
5 years ago
5 years ago
5 years ago
5 years ago
5 years ago
5 years ago
5 years ago
5 years ago
5 years ago
  1. package weed_server
  2. import (
  3. "context"
  4. "fmt"
  5. "io"
  6. "math"
  7. "os"
  8. "path"
  9. "strings"
  10. "time"
  11. "github.com/chrislusf/seaweedfs/weed/util/grace"
  12. "golang.org/x/net/webdav"
  13. "google.golang.org/grpc"
  14. "github.com/chrislusf/seaweedfs/weed/operation"
  15. "github.com/chrislusf/seaweedfs/weed/pb"
  16. "github.com/chrislusf/seaweedfs/weed/pb/filer_pb"
  17. "github.com/chrislusf/seaweedfs/weed/util"
  18. "github.com/chrislusf/seaweedfs/weed/util/chunk_cache"
  19. "github.com/chrislusf/seaweedfs/weed/filer2"
  20. "github.com/chrislusf/seaweedfs/weed/glog"
  21. "github.com/chrislusf/seaweedfs/weed/security"
  22. )
  23. type WebDavOption struct {
  24. Filer string
  25. FilerGrpcAddress string
  26. DomainName string
  27. BucketsPath string
  28. GrpcDialOption grpc.DialOption
  29. Collection string
  30. Uid uint32
  31. Gid uint32
  32. Cipher bool
  33. CacheDir string
  34. CacheSizeMB int64
  35. }
  36. type WebDavServer struct {
  37. option *WebDavOption
  38. secret security.SigningKey
  39. filer *filer2.Filer
  40. grpcDialOption grpc.DialOption
  41. Handler *webdav.Handler
  42. }
  43. func NewWebDavServer(option *WebDavOption) (ws *WebDavServer, err error) {
  44. fs, _ := NewWebDavFileSystem(option)
  45. ws = &WebDavServer{
  46. option: option,
  47. grpcDialOption: security.LoadClientTLS(util.GetViper(), "grpc.filer"),
  48. Handler: &webdav.Handler{
  49. FileSystem: fs,
  50. LockSystem: webdav.NewMemLS(),
  51. },
  52. }
  53. return ws, nil
  54. }
  55. // adapted from https://github.com/mattn/davfs/blob/master/plugin/mysql/mysql.go
  56. type WebDavFileSystem struct {
  57. option *WebDavOption
  58. secret security.SigningKey
  59. filer *filer2.Filer
  60. grpcDialOption grpc.DialOption
  61. chunkCache *chunk_cache.ChunkCache
  62. }
  63. type FileInfo struct {
  64. name string
  65. size int64
  66. mode os.FileMode
  67. modifiledTime time.Time
  68. isDirectory bool
  69. }
  70. func (fi *FileInfo) Name() string { return fi.name }
  71. func (fi *FileInfo) Size() int64 { return fi.size }
  72. func (fi *FileInfo) Mode() os.FileMode { return fi.mode }
  73. func (fi *FileInfo) ModTime() time.Time { return fi.modifiledTime }
  74. func (fi *FileInfo) IsDir() bool { return fi.isDirectory }
  75. func (fi *FileInfo) Sys() interface{} { return nil }
  76. type WebDavFile struct {
  77. fs *WebDavFileSystem
  78. name string
  79. isDirectory bool
  80. off int64
  81. entry *filer_pb.Entry
  82. entryViewCache []filer2.VisibleInterval
  83. reader io.ReaderAt
  84. }
  85. func NewWebDavFileSystem(option *WebDavOption) (webdav.FileSystem, error) {
  86. chunkCache := chunk_cache.NewChunkCache(256, option.CacheDir, option.CacheSizeMB)
  87. grace.OnInterrupt(func() {
  88. chunkCache.Shutdown()
  89. })
  90. return &WebDavFileSystem{
  91. option: option,
  92. chunkCache: chunkCache,
  93. }, nil
  94. }
  95. var _ = filer_pb.FilerClient(&WebDavFileSystem{})
  96. func (fs *WebDavFileSystem) WithFilerClient(fn func(filer_pb.SeaweedFilerClient) error) error {
  97. return pb.WithCachedGrpcClient(func(grpcConnection *grpc.ClientConn) error {
  98. client := filer_pb.NewSeaweedFilerClient(grpcConnection)
  99. return fn(client)
  100. }, fs.option.FilerGrpcAddress, fs.option.GrpcDialOption)
  101. }
  102. func (fs *WebDavFileSystem) AdjustedUrl(hostAndPort string) string {
  103. return hostAndPort
  104. }
  105. func clearName(name string) (string, error) {
  106. slashed := strings.HasSuffix(name, "/")
  107. name = path.Clean(name)
  108. if !strings.HasSuffix(name, "/") && slashed {
  109. name += "/"
  110. }
  111. if !strings.HasPrefix(name, "/") {
  112. return "", os.ErrInvalid
  113. }
  114. return name, nil
  115. }
  116. func (fs *WebDavFileSystem) Mkdir(ctx context.Context, fullDirPath string, perm os.FileMode) error {
  117. glog.V(2).Infof("WebDavFileSystem.Mkdir %v", fullDirPath)
  118. if !strings.HasSuffix(fullDirPath, "/") {
  119. fullDirPath += "/"
  120. }
  121. var err error
  122. if fullDirPath, err = clearName(fullDirPath); err != nil {
  123. return err
  124. }
  125. _, err = fs.stat(ctx, fullDirPath)
  126. if err == nil {
  127. return os.ErrExist
  128. }
  129. return fs.WithFilerClient(func(client filer_pb.SeaweedFilerClient) error {
  130. dir, name := util.FullPath(fullDirPath).DirAndName()
  131. request := &filer_pb.CreateEntryRequest{
  132. Directory: dir,
  133. Entry: &filer_pb.Entry{
  134. Name: name,
  135. IsDirectory: true,
  136. Attributes: &filer_pb.FuseAttributes{
  137. Mtime: time.Now().Unix(),
  138. Crtime: time.Now().Unix(),
  139. FileMode: uint32(perm | os.ModeDir),
  140. Uid: fs.option.Uid,
  141. Gid: fs.option.Gid,
  142. },
  143. },
  144. }
  145. glog.V(1).Infof("mkdir: %v", request)
  146. if err := filer_pb.CreateEntry(client, request); err != nil {
  147. return fmt.Errorf("mkdir %s/%s: %v", dir, name, err)
  148. }
  149. return nil
  150. })
  151. }
  152. func (fs *WebDavFileSystem) OpenFile(ctx context.Context, fullFilePath string, flag int, perm os.FileMode) (webdav.File, error) {
  153. glog.V(2).Infof("WebDavFileSystem.OpenFile %v %x", fullFilePath, flag)
  154. var err error
  155. if fullFilePath, err = clearName(fullFilePath); err != nil {
  156. return nil, err
  157. }
  158. if flag&os.O_CREATE != 0 {
  159. // file should not have / suffix.
  160. if strings.HasSuffix(fullFilePath, "/") {
  161. return nil, os.ErrInvalid
  162. }
  163. _, err = fs.stat(ctx, fullFilePath)
  164. if err == nil {
  165. if flag&os.O_EXCL != 0 {
  166. return nil, os.ErrExist
  167. }
  168. fs.removeAll(ctx, fullFilePath)
  169. }
  170. dir, name := util.FullPath(fullFilePath).DirAndName()
  171. err = fs.WithFilerClient(func(client filer_pb.SeaweedFilerClient) error {
  172. if err := filer_pb.CreateEntry(client, &filer_pb.CreateEntryRequest{
  173. Directory: dir,
  174. Entry: &filer_pb.Entry{
  175. Name: name,
  176. IsDirectory: perm&os.ModeDir > 0,
  177. Attributes: &filer_pb.FuseAttributes{
  178. Mtime: time.Now().Unix(),
  179. Crtime: time.Now().Unix(),
  180. FileMode: uint32(perm),
  181. Uid: fs.option.Uid,
  182. Gid: fs.option.Gid,
  183. Collection: fs.option.Collection,
  184. Replication: "000",
  185. TtlSec: 0,
  186. },
  187. },
  188. }); err != nil {
  189. return fmt.Errorf("create %s: %v", fullFilePath, err)
  190. }
  191. return nil
  192. })
  193. if err != nil {
  194. return nil, err
  195. }
  196. return &WebDavFile{
  197. fs: fs,
  198. name: fullFilePath,
  199. isDirectory: false,
  200. }, nil
  201. }
  202. fi, err := fs.stat(ctx, fullFilePath)
  203. if err != nil {
  204. return nil, os.ErrNotExist
  205. }
  206. if !strings.HasSuffix(fullFilePath, "/") && fi.IsDir() {
  207. fullFilePath += "/"
  208. }
  209. return &WebDavFile{
  210. fs: fs,
  211. name: fullFilePath,
  212. isDirectory: false,
  213. }, nil
  214. }
  215. func (fs *WebDavFileSystem) removeAll(ctx context.Context, fullFilePath string) error {
  216. var err error
  217. if fullFilePath, err = clearName(fullFilePath); err != nil {
  218. return err
  219. }
  220. dir, name := util.FullPath(fullFilePath).DirAndName()
  221. return filer_pb.Remove(fs, dir, name, true, false, false)
  222. }
  223. func (fs *WebDavFileSystem) RemoveAll(ctx context.Context, name string) error {
  224. glog.V(2).Infof("WebDavFileSystem.RemoveAll %v", name)
  225. return fs.removeAll(ctx, name)
  226. }
  227. func (fs *WebDavFileSystem) Rename(ctx context.Context, oldName, newName string) error {
  228. glog.V(2).Infof("WebDavFileSystem.Rename %v to %v", oldName, newName)
  229. var err error
  230. if oldName, err = clearName(oldName); err != nil {
  231. return err
  232. }
  233. if newName, err = clearName(newName); err != nil {
  234. return err
  235. }
  236. of, err := fs.stat(ctx, oldName)
  237. if err != nil {
  238. return os.ErrExist
  239. }
  240. if of.IsDir() {
  241. if strings.HasSuffix(oldName, "/") {
  242. oldName = strings.TrimRight(oldName, "/")
  243. }
  244. if strings.HasSuffix(newName, "/") {
  245. newName = strings.TrimRight(newName, "/")
  246. }
  247. }
  248. _, err = fs.stat(ctx, newName)
  249. if err == nil {
  250. return os.ErrExist
  251. }
  252. oldDir, oldBaseName := util.FullPath(oldName).DirAndName()
  253. newDir, newBaseName := util.FullPath(newName).DirAndName()
  254. return fs.WithFilerClient(func(client filer_pb.SeaweedFilerClient) error {
  255. request := &filer_pb.AtomicRenameEntryRequest{
  256. OldDirectory: oldDir,
  257. OldName: oldBaseName,
  258. NewDirectory: newDir,
  259. NewName: newBaseName,
  260. }
  261. _, err := client.AtomicRenameEntry(ctx, request)
  262. if err != nil {
  263. return fmt.Errorf("renaming %s/%s => %s/%s: %v", oldDir, oldBaseName, newDir, newBaseName, err)
  264. }
  265. return nil
  266. })
  267. }
  268. func (fs *WebDavFileSystem) stat(ctx context.Context, fullFilePath string) (os.FileInfo, error) {
  269. var err error
  270. if fullFilePath, err = clearName(fullFilePath); err != nil {
  271. return nil, err
  272. }
  273. fullpath := util.FullPath(fullFilePath)
  274. var fi FileInfo
  275. entry, err := filer_pb.GetEntry(fs, fullpath)
  276. if entry == nil {
  277. return nil, os.ErrNotExist
  278. }
  279. if err != nil {
  280. return nil, err
  281. }
  282. fi.size = int64(filer2.TotalSize(entry.GetChunks()))
  283. fi.name = string(fullpath)
  284. fi.mode = os.FileMode(entry.Attributes.FileMode)
  285. fi.modifiledTime = time.Unix(entry.Attributes.Mtime, 0)
  286. fi.isDirectory = entry.IsDirectory
  287. if fi.name == "/" {
  288. fi.modifiledTime = time.Now()
  289. fi.isDirectory = true
  290. }
  291. return &fi, nil
  292. }
  293. func (fs *WebDavFileSystem) Stat(ctx context.Context, name string) (os.FileInfo, error) {
  294. glog.V(2).Infof("WebDavFileSystem.Stat %v", name)
  295. return fs.stat(ctx, name)
  296. }
  297. func (f *WebDavFile) Write(buf []byte) (int, error) {
  298. glog.V(2).Infof("WebDavFileSystem.Write %v", f.name)
  299. dir, _ := util.FullPath(f.name).DirAndName()
  300. var err error
  301. ctx := context.Background()
  302. if f.entry == nil {
  303. f.entry, err = filer_pb.GetEntry(f.fs, util.FullPath(f.name))
  304. }
  305. if f.entry == nil {
  306. return 0, err
  307. }
  308. if err != nil {
  309. return 0, err
  310. }
  311. var fileId, host string
  312. var auth security.EncodedJwt
  313. var collection, replication string
  314. if err = f.fs.WithFilerClient(func(client filer_pb.SeaweedFilerClient) error {
  315. request := &filer_pb.AssignVolumeRequest{
  316. Count: 1,
  317. Replication: "",
  318. Collection: f.fs.option.Collection,
  319. ParentPath: dir,
  320. }
  321. resp, err := client.AssignVolume(ctx, request)
  322. if err != nil {
  323. glog.V(0).Infof("assign volume failure %v: %v", request, err)
  324. return err
  325. }
  326. if resp.Error != "" {
  327. return fmt.Errorf("assign volume failure %v: %v", request, resp.Error)
  328. }
  329. fileId, host, auth = resp.FileId, resp.Url, security.EncodedJwt(resp.Auth)
  330. collection, replication = resp.Collection, resp.Replication
  331. return nil
  332. }); err != nil {
  333. return 0, fmt.Errorf("filerGrpcAddress assign volume: %v", err)
  334. }
  335. fileUrl := fmt.Sprintf("http://%s/%s", host, fileId)
  336. uploadResult, err := operation.UploadData(fileUrl, f.name, f.fs.option.Cipher, buf, false, "", nil, auth)
  337. if err != nil {
  338. glog.V(0).Infof("upload data %v to %s: %v", f.name, fileUrl, err)
  339. return 0, fmt.Errorf("upload data: %v", err)
  340. }
  341. if uploadResult.Error != "" {
  342. glog.V(0).Infof("upload failure %v to %s: %v", f.name, fileUrl, err)
  343. return 0, fmt.Errorf("upload result: %v", uploadResult.Error)
  344. }
  345. f.entry.Chunks = append(f.entry.Chunks, uploadResult.ToPbFileChunk(fileId, f.off))
  346. err = f.fs.WithFilerClient(func(client filer_pb.SeaweedFilerClient) error {
  347. f.entry.Attributes.Mtime = time.Now().Unix()
  348. f.entry.Attributes.Collection = collection
  349. f.entry.Attributes.Replication = replication
  350. request := &filer_pb.UpdateEntryRequest{
  351. Directory: dir,
  352. Entry: f.entry,
  353. }
  354. if _, err := client.UpdateEntry(ctx, request); err != nil {
  355. return fmt.Errorf("update %s: %v", f.name, err)
  356. }
  357. return nil
  358. })
  359. if err == nil {
  360. glog.V(3).Infof("WebDavFileSystem.Write %v: written [%d,%d)", f.name, f.off, f.off+int64(len(buf)))
  361. f.off += int64(len(buf))
  362. }
  363. return len(buf), err
  364. }
  365. func (f *WebDavFile) Close() error {
  366. glog.V(2).Infof("WebDavFileSystem.Close %v", f.name)
  367. if f.entry != nil {
  368. f.entry = nil
  369. f.entryViewCache = nil
  370. }
  371. return nil
  372. }
  373. func (f *WebDavFile) Read(p []byte) (readSize int, err error) {
  374. glog.V(2).Infof("WebDavFileSystem.Read %v", f.name)
  375. if f.entry == nil {
  376. f.entry, err = filer_pb.GetEntry(f.fs, util.FullPath(f.name))
  377. }
  378. if f.entry == nil {
  379. return 0, err
  380. }
  381. if err != nil {
  382. return 0, err
  383. }
  384. if len(f.entry.Chunks) == 0 {
  385. return 0, io.EOF
  386. }
  387. if f.entryViewCache == nil {
  388. f.entryViewCache = filer2.NonOverlappingVisibleIntervals(f.entry.Chunks)
  389. f.reader = nil
  390. }
  391. if f.reader == nil {
  392. chunkViews := filer2.ViewFromVisibleIntervals(f.entryViewCache, 0, math.MaxInt32)
  393. f.reader = filer2.NewChunkReaderAtFromClient(f.fs, chunkViews, f.fs.chunkCache)
  394. }
  395. readSize, err = f.reader.ReadAt(p, f.off)
  396. glog.V(3).Infof("WebDavFileSystem.Read %v: [%d,%d)", f.name, f.off, f.off+int64(readSize))
  397. f.off += int64(readSize)
  398. if err != nil {
  399. glog.Errorf("file read %s: %v", f.name, err)
  400. }
  401. return
  402. }
  403. func (f *WebDavFile) Readdir(count int) (ret []os.FileInfo, err error) {
  404. glog.V(2).Infof("WebDavFileSystem.Readdir %v count %d", f.name, count)
  405. dir, _ := util.FullPath(f.name).DirAndName()
  406. err = filer_pb.ReadDirAllEntries(f.fs, util.FullPath(dir), "", func(entry *filer_pb.Entry, isLast bool) error {
  407. fi := FileInfo{
  408. size: int64(filer2.TotalSize(entry.GetChunks())),
  409. name: entry.Name,
  410. mode: os.FileMode(entry.Attributes.FileMode),
  411. modifiledTime: time.Unix(entry.Attributes.Mtime, 0),
  412. isDirectory: entry.IsDirectory,
  413. }
  414. if !strings.HasSuffix(fi.name, "/") && fi.IsDir() {
  415. fi.name += "/"
  416. }
  417. glog.V(4).Infof("entry: %v", fi.name)
  418. ret = append(ret, &fi)
  419. return nil
  420. })
  421. old := f.off
  422. if old >= int64(len(ret)) {
  423. if count > 0 {
  424. return nil, io.EOF
  425. }
  426. return nil, nil
  427. }
  428. if count > 0 {
  429. f.off += int64(count)
  430. if f.off > int64(len(ret)) {
  431. f.off = int64(len(ret))
  432. }
  433. } else {
  434. f.off = int64(len(ret))
  435. old = 0
  436. }
  437. return ret[old:f.off], nil
  438. }
  439. func (f *WebDavFile) Seek(offset int64, whence int) (int64, error) {
  440. glog.V(2).Infof("WebDavFile.Seek %v %v %v", f.name, offset, whence)
  441. ctx := context.Background()
  442. var err error
  443. switch whence {
  444. case 0:
  445. f.off = 0
  446. case 2:
  447. if fi, err := f.fs.stat(ctx, f.name); err != nil {
  448. return 0, err
  449. } else {
  450. f.off = fi.Size()
  451. }
  452. }
  453. f.off += offset
  454. return f.off, err
  455. }
  456. func (f *WebDavFile) Stat() (os.FileInfo, error) {
  457. glog.V(2).Infof("WebDavFile.Stat %v", f.name)
  458. ctx := context.Background()
  459. return f.fs.stat(ctx, f.name)
  460. }