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.

578 lines
14 KiB

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