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.

323 lines
7.8 KiB

7 years ago
7 years ago
7 years ago
7 years ago
7 years ago
7 years ago
  1. package filesys
  2. import (
  3. "context"
  4. "os"
  5. "path"
  6. "path/filepath"
  7. "time"
  8. "bazil.org/fuse"
  9. "bazil.org/fuse/fs"
  10. "github.com/chrislusf/seaweedfs/weed/filer2"
  11. "github.com/chrislusf/seaweedfs/weed/glog"
  12. "github.com/chrislusf/seaweedfs/weed/pb/filer_pb"
  13. )
  14. type Dir struct {
  15. Path string
  16. wfs *WFS
  17. attributes *filer_pb.FuseAttributes
  18. }
  19. var _ = fs.Node(&Dir{})
  20. var _ = fs.NodeCreater(&Dir{})
  21. var _ = fs.NodeMkdirer(&Dir{})
  22. var _ = fs.NodeRequestLookuper(&Dir{})
  23. var _ = fs.HandleReadDirAller(&Dir{})
  24. var _ = fs.NodeRemover(&Dir{})
  25. var _ = fs.NodeRenamer(&Dir{})
  26. var _ = fs.NodeSetattrer(&Dir{})
  27. func (dir *Dir) Attr(context context.Context, attr *fuse.Attr) error {
  28. if dir.Path == "/" {
  29. attr.Valid = time.Second
  30. attr.Mode = os.ModeDir | 0777
  31. return nil
  32. }
  33. item := dir.wfs.listDirectoryEntriesCache.Get(dir.Path)
  34. if item != nil && !item.Expired() {
  35. entry := item.Value().(*filer_pb.Entry)
  36. attr.Mtime = time.Unix(entry.Attributes.Mtime, 0)
  37. attr.Ctime = time.Unix(entry.Attributes.Crtime, 0)
  38. attr.Mode = os.FileMode(entry.Attributes.FileMode)
  39. attr.Gid = entry.Attributes.Gid
  40. attr.Uid = entry.Attributes.Uid
  41. return nil
  42. }
  43. parent, name := filepath.Split(dir.Path)
  44. err := dir.wfs.withFilerClient(func(client filer_pb.SeaweedFilerClient) error {
  45. request := &filer_pb.LookupDirectoryEntryRequest{
  46. Directory: parent,
  47. Name: name,
  48. }
  49. glog.V(1).Infof("read dir %s attr: %v", dir.Path, request)
  50. resp, err := client.LookupDirectoryEntry(context, request)
  51. if err != nil {
  52. glog.V(0).Infof("read dir %s attr %v: %v", dir.Path, request, err)
  53. return err
  54. }
  55. if resp.Entry!=nil {
  56. dir.attributes = resp.Entry.Attributes
  57. }
  58. return nil
  59. })
  60. if err != nil {
  61. return err
  62. }
  63. // glog.V(1).Infof("dir %s: %v", dir.Path, attributes)
  64. // glog.V(1).Infof("dir %s permission: %v", dir.Path, os.FileMode(attributes.FileMode))
  65. attr.Mode = os.FileMode(dir.attributes.FileMode) | os.ModeDir
  66. if dir.Path == "/" && dir.attributes.FileMode == 0 {
  67. attr.Valid = time.Second
  68. }
  69. attr.Mtime = time.Unix(dir.attributes.Mtime, 0)
  70. attr.Ctime = time.Unix(dir.attributes.Crtime, 0)
  71. attr.Gid = dir.attributes.Gid
  72. attr.Uid = dir.attributes.Uid
  73. return nil
  74. }
  75. func (dir *Dir) newFile(name string, entry *filer_pb.Entry) *File {
  76. return &File{
  77. Name: name,
  78. dir: dir,
  79. wfs: dir.wfs,
  80. entry: entry,
  81. }
  82. }
  83. func (dir *Dir) Create(ctx context.Context, req *fuse.CreateRequest,
  84. resp *fuse.CreateResponse) (fs.Node, fs.Handle, error) {
  85. request := &filer_pb.CreateEntryRequest{
  86. Directory: dir.Path,
  87. Entry: &filer_pb.Entry{
  88. Name: req.Name,
  89. IsDirectory: req.Mode&os.ModeDir > 0,
  90. Attributes: &filer_pb.FuseAttributes{
  91. Mtime: time.Now().Unix(),
  92. Crtime: time.Now().Unix(),
  93. FileMode: uint32(req.Mode),
  94. Uid: req.Uid,
  95. Gid: req.Gid,
  96. Collection: dir.wfs.option.Collection,
  97. Replication: dir.wfs.option.Replication,
  98. TtlSec: dir.wfs.option.TtlSec,
  99. },
  100. },
  101. }
  102. glog.V(1).Infof("create: %v", request)
  103. if request.Entry.IsDirectory {
  104. if err := dir.wfs.withFilerClient(func(client filer_pb.SeaweedFilerClient) error {
  105. if _, err := client.CreateEntry(ctx, request); err != nil {
  106. glog.V(0).Infof("create %s/%s: %v", dir.Path, req.Name, err)
  107. return fuse.EIO
  108. }
  109. return nil
  110. }); err != nil {
  111. return nil, nil, err
  112. }
  113. }
  114. file := dir.newFile(req.Name, request.Entry)
  115. if !request.Entry.IsDirectory {
  116. file.isOpen = true
  117. }
  118. fh := dir.wfs.AcquireHandle(file, req.Uid, req.Gid)
  119. fh.dirtyMetadata = true
  120. return file, fh, nil
  121. }
  122. func (dir *Dir) Mkdir(ctx context.Context, req *fuse.MkdirRequest) (fs.Node, error) {
  123. err := dir.wfs.withFilerClient(func(client filer_pb.SeaweedFilerClient) error {
  124. request := &filer_pb.CreateEntryRequest{
  125. Directory: dir.Path,
  126. Entry: &filer_pb.Entry{
  127. Name: req.Name,
  128. IsDirectory: true,
  129. Attributes: &filer_pb.FuseAttributes{
  130. Mtime: time.Now().Unix(),
  131. Crtime: time.Now().Unix(),
  132. FileMode: uint32(req.Mode),
  133. Uid: req.Uid,
  134. Gid: req.Gid,
  135. },
  136. },
  137. }
  138. glog.V(1).Infof("mkdir: %v", request)
  139. if _, err := client.CreateEntry(ctx, request); err != nil {
  140. glog.V(0).Infof("mkdir %s/%s: %v", dir.Path, req.Name, err)
  141. return fuse.EIO
  142. }
  143. return nil
  144. })
  145. if err == nil {
  146. node := &Dir{Path: path.Join(dir.Path, req.Name), wfs: dir.wfs}
  147. return node, nil
  148. }
  149. return nil, err
  150. }
  151. func (dir *Dir) Lookup(ctx context.Context, req *fuse.LookupRequest, resp *fuse.LookupResponse) (node fs.Node, err error) {
  152. var entry *filer_pb.Entry
  153. err = dir.wfs.withFilerClient(func(client filer_pb.SeaweedFilerClient) error {
  154. request := &filer_pb.LookupDirectoryEntryRequest{
  155. Directory: dir.Path,
  156. Name: req.Name,
  157. }
  158. glog.V(4).Infof("lookup directory entry: %v", request)
  159. resp, err := client.LookupDirectoryEntry(ctx, request)
  160. if err != nil {
  161. // glog.V(0).Infof("lookup %s/%s: %v", dir.Path, name, err)
  162. return fuse.ENOENT
  163. }
  164. entry = resp.Entry
  165. return nil
  166. })
  167. if entry != nil {
  168. if entry.IsDirectory {
  169. node = &Dir{Path: path.Join(dir.Path, req.Name), wfs: dir.wfs, attributes: entry.Attributes}
  170. } else {
  171. node = dir.newFile(req.Name, entry)
  172. }
  173. resp.EntryValid = time.Duration(0)
  174. resp.Attr.Mtime = time.Unix(entry.Attributes.Mtime, 0)
  175. resp.Attr.Ctime = time.Unix(entry.Attributes.Crtime, 0)
  176. resp.Attr.Mode = os.FileMode(entry.Attributes.FileMode)
  177. resp.Attr.Gid = entry.Attributes.Gid
  178. resp.Attr.Uid = entry.Attributes.Uid
  179. return node, nil
  180. }
  181. return nil, fuse.ENOENT
  182. }
  183. func (dir *Dir) ReadDirAll(ctx context.Context) (ret []fuse.Dirent, err error) {
  184. err = dir.wfs.withFilerClient(func(client filer_pb.SeaweedFilerClient) error {
  185. request := &filer_pb.ListEntriesRequest{
  186. Directory: dir.Path,
  187. Limit: uint32(dir.wfs.option.DirListingLimit),
  188. }
  189. glog.V(4).Infof("read directory: %v", request)
  190. resp, err := client.ListEntries(ctx, request)
  191. if err != nil {
  192. glog.V(0).Infof("list %s: %v", dir.Path, err)
  193. return fuse.EIO
  194. }
  195. for _, entry := range resp.Entries {
  196. if entry.IsDirectory {
  197. dirent := fuse.Dirent{Name: entry.Name, Type: fuse.DT_Dir}
  198. ret = append(ret, dirent)
  199. } else {
  200. dirent := fuse.Dirent{Name: entry.Name, Type: fuse.DT_File}
  201. ret = append(ret, dirent)
  202. }
  203. dir.wfs.listDirectoryEntriesCache.Set(dir.Path+"/"+entry.Name, entry, 300*time.Millisecond)
  204. }
  205. return nil
  206. })
  207. return ret, err
  208. }
  209. func (dir *Dir) Remove(ctx context.Context, req *fuse.RemoveRequest) error {
  210. return dir.wfs.withFilerClient(func(client filer_pb.SeaweedFilerClient) error {
  211. request := &filer_pb.DeleteEntryRequest{
  212. Directory: dir.Path,
  213. Name: req.Name,
  214. IsDirectory: req.Dir,
  215. IsDeleteData: true,
  216. }
  217. glog.V(1).Infof("remove directory entry: %v", request)
  218. _, err := client.DeleteEntry(ctx, request)
  219. if err != nil {
  220. glog.V(0).Infof("remove %s/%s: %v", dir.Path, req.Name, err)
  221. return fuse.EIO
  222. }
  223. return nil
  224. })
  225. }
  226. func (dir *Dir) Setattr(ctx context.Context, req *fuse.SetattrRequest, resp *fuse.SetattrResponse) error {
  227. glog.V(3).Infof("%v dir setattr %+v, fh=%d", dir.Path, req, req.Handle)
  228. if req.Valid.Mode() {
  229. dir.attributes.FileMode = uint32(req.Mode)
  230. }
  231. if req.Valid.Uid() {
  232. dir.attributes.Uid = req.Uid
  233. }
  234. if req.Valid.Gid() {
  235. dir.attributes.Gid = req.Gid
  236. }
  237. if req.Valid.Mtime() {
  238. dir.attributes.Mtime = req.Mtime.Unix()
  239. }
  240. parentDir, name := filer2.FullPath(dir.Path).DirAndName()
  241. return dir.wfs.withFilerClient(func(client filer_pb.SeaweedFilerClient) error {
  242. request := &filer_pb.UpdateEntryRequest{
  243. Directory: parentDir,
  244. Entry: &filer_pb.Entry{
  245. Name: name,
  246. Attributes: dir.attributes,
  247. },
  248. }
  249. glog.V(1).Infof("set attr directory entry: %v", request)
  250. _, err := client.UpdateEntry(ctx, request)
  251. if err != nil {
  252. glog.V(0).Infof("UpdateEntry %s: %v", dir.Path, err)
  253. return fuse.EIO
  254. }
  255. return nil
  256. })
  257. }