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.

261 lines
6.9 KiB

5 years ago
7 years ago
5 years ago
5 years ago
6 years ago
5 years ago
6 years ago
5 years ago
5 years ago
6 years ago
5 years ago
5 years ago
  1. package filesys
  2. import (
  3. "context"
  4. "fmt"
  5. "math"
  6. "os"
  7. "path"
  8. "strings"
  9. "sync"
  10. "time"
  11. "github.com/karlseguin/ccache"
  12. "google.golang.org/grpc"
  13. "github.com/chrislusf/seaweedfs/weed/util/grace"
  14. "github.com/seaweedfs/fuse"
  15. "github.com/seaweedfs/fuse/fs"
  16. "github.com/chrislusf/seaweedfs/weed/filesys/meta_cache"
  17. "github.com/chrislusf/seaweedfs/weed/glog"
  18. "github.com/chrislusf/seaweedfs/weed/pb"
  19. "github.com/chrislusf/seaweedfs/weed/pb/filer_pb"
  20. "github.com/chrislusf/seaweedfs/weed/util"
  21. "github.com/chrislusf/seaweedfs/weed/util/chunk_cache"
  22. )
  23. type Option struct {
  24. FilerGrpcAddress string
  25. GrpcDialOption grpc.DialOption
  26. FilerMountRootPath string
  27. Collection string
  28. Replication string
  29. TtlSec int32
  30. ChunkSizeLimit int64
  31. CacheDir string
  32. CacheSizeMB int64
  33. DataCenter string
  34. DirListCacheLimit int64
  35. EntryCacheTtl time.Duration
  36. Umask os.FileMode
  37. MountUid uint32
  38. MountGid uint32
  39. MountMode os.FileMode
  40. MountCtime time.Time
  41. MountMtime time.Time
  42. OutsideContainerClusterMode bool // whether the mount runs outside SeaweedFS containers
  43. Cipher bool // whether encrypt data on volume server
  44. }
  45. var _ = fs.FS(&WFS{})
  46. var _ = fs.FSStatfser(&WFS{})
  47. type WFS struct {
  48. option *Option
  49. listDirectoryEntriesCache *ccache.Cache
  50. // contains all open handles, protected by handlesLock
  51. handlesLock sync.Mutex
  52. handles map[uint64]*FileHandle
  53. bufPool sync.Pool
  54. stats statsCache
  55. root fs.Node
  56. fsNodeCache *FsCache
  57. chunkCache *chunk_cache.ChunkCache
  58. metaCache *meta_cache.MetaCache
  59. }
  60. type statsCache struct {
  61. filer_pb.StatisticsResponse
  62. lastChecked int64 // unix time in seconds
  63. }
  64. func NewSeaweedFileSystem(option *Option) *WFS {
  65. wfs := &WFS{
  66. option: option,
  67. listDirectoryEntriesCache: ccache.New(ccache.Configure().MaxSize(option.DirListCacheLimit * 3).ItemsToPrune(100)),
  68. handles: make(map[uint64]*FileHandle),
  69. bufPool: sync.Pool{
  70. New: func() interface{} {
  71. return make([]byte, option.ChunkSizeLimit)
  72. },
  73. },
  74. }
  75. if option.CacheSizeMB > 0 {
  76. os.MkdirAll(option.CacheDir, 0755)
  77. wfs.chunkCache = chunk_cache.NewChunkCache(256, option.CacheDir, option.CacheSizeMB)
  78. grace.OnInterrupt(func() {
  79. wfs.chunkCache.Shutdown()
  80. })
  81. }
  82. wfs.metaCache = meta_cache.NewMetaCache(path.Join(option.CacheDir, "meta"))
  83. startTime := time.Now()
  84. if err := meta_cache.InitMetaCache(wfs.metaCache, wfs, wfs.option.FilerMountRootPath); err != nil {
  85. glog.V(0).Infof("failed to init meta cache: %v", err)
  86. } else {
  87. go meta_cache.SubscribeMetaEvents(wfs.metaCache, wfs, wfs.option.FilerMountRootPath, startTime.UnixNano())
  88. grace.OnInterrupt(func() {
  89. wfs.metaCache.Shutdown()
  90. })
  91. }
  92. wfs.root = &Dir{name: wfs.option.FilerMountRootPath, wfs: wfs}
  93. wfs.fsNodeCache = newFsCache(wfs.root)
  94. return wfs
  95. }
  96. func (wfs *WFS) Root() (fs.Node, error) {
  97. return wfs.root, nil
  98. }
  99. var _ = filer_pb.FilerClient(&WFS{})
  100. func (wfs *WFS) WithFilerClient(fn func(filer_pb.SeaweedFilerClient) error) error {
  101. err := pb.WithCachedGrpcClient(func(grpcConnection *grpc.ClientConn) error {
  102. client := filer_pb.NewSeaweedFilerClient(grpcConnection)
  103. return fn(client)
  104. }, wfs.option.FilerGrpcAddress, wfs.option.GrpcDialOption)
  105. if err == nil {
  106. return nil
  107. }
  108. return err
  109. }
  110. func (wfs *WFS) AcquireHandle(file *File, uid, gid uint32) (fileHandle *FileHandle) {
  111. fullpath := file.fullpath()
  112. glog.V(4).Infof("%s AcquireHandle uid=%d gid=%d", fullpath, uid, gid)
  113. wfs.handlesLock.Lock()
  114. defer wfs.handlesLock.Unlock()
  115. inodeId := file.fullpath().AsInode()
  116. existingHandle, found := wfs.handles[inodeId]
  117. if found && existingHandle != nil {
  118. return existingHandle
  119. }
  120. fileHandle = newFileHandle(file, uid, gid)
  121. wfs.handles[inodeId] = fileHandle
  122. fileHandle.handle = inodeId
  123. glog.V(4).Infof("%s new fh %d", fullpath, fileHandle.handle)
  124. return
  125. }
  126. func (wfs *WFS) ReleaseHandle(fullpath util.FullPath, handleId fuse.HandleID) {
  127. wfs.handlesLock.Lock()
  128. defer wfs.handlesLock.Unlock()
  129. glog.V(4).Infof("%s ReleaseHandle id %d current handles length %d", fullpath, handleId, len(wfs.handles))
  130. delete(wfs.handles, fullpath.AsInode())
  131. return
  132. }
  133. // Statfs is called to obtain file system metadata. Implements fuse.FSStatfser
  134. func (wfs *WFS) Statfs(ctx context.Context, req *fuse.StatfsRequest, resp *fuse.StatfsResponse) error {
  135. glog.V(4).Infof("reading fs stats: %+v", req)
  136. if wfs.stats.lastChecked < time.Now().Unix()-20 {
  137. err := wfs.WithFilerClient(func(client filer_pb.SeaweedFilerClient) error {
  138. request := &filer_pb.StatisticsRequest{
  139. Collection: wfs.option.Collection,
  140. Replication: wfs.option.Replication,
  141. Ttl: fmt.Sprintf("%ds", wfs.option.TtlSec),
  142. }
  143. glog.V(4).Infof("reading filer stats: %+v", request)
  144. resp, err := client.Statistics(context.Background(), request)
  145. if err != nil {
  146. glog.V(0).Infof("reading filer stats %v: %v", request, err)
  147. return err
  148. }
  149. glog.V(4).Infof("read filer stats: %+v", resp)
  150. wfs.stats.TotalSize = resp.TotalSize
  151. wfs.stats.UsedSize = resp.UsedSize
  152. wfs.stats.FileCount = resp.FileCount
  153. wfs.stats.lastChecked = time.Now().Unix()
  154. return nil
  155. })
  156. if err != nil {
  157. glog.V(0).Infof("filer Statistics: %v", err)
  158. return err
  159. }
  160. }
  161. totalDiskSize := wfs.stats.TotalSize
  162. usedDiskSize := wfs.stats.UsedSize
  163. actualFileCount := wfs.stats.FileCount
  164. // Compute the total number of available blocks
  165. resp.Blocks = totalDiskSize / blockSize
  166. // Compute the number of used blocks
  167. numBlocks := uint64(usedDiskSize / blockSize)
  168. // Report the number of free and available blocks for the block size
  169. resp.Bfree = resp.Blocks - numBlocks
  170. resp.Bavail = resp.Blocks - numBlocks
  171. resp.Bsize = uint32(blockSize)
  172. // Report the total number of possible files in the file system (and those free)
  173. resp.Files = math.MaxInt64
  174. resp.Ffree = math.MaxInt64 - actualFileCount
  175. // Report the maximum length of a name and the minimum fragment size
  176. resp.Namelen = 1024
  177. resp.Frsize = uint32(blockSize)
  178. return nil
  179. }
  180. func (wfs *WFS) cacheGet(path util.FullPath) *filer_pb.Entry {
  181. item := wfs.listDirectoryEntriesCache.Get(string(path))
  182. if item != nil && !item.Expired() {
  183. return item.Value().(*filer_pb.Entry)
  184. }
  185. return nil
  186. }
  187. func (wfs *WFS) cacheSet(path util.FullPath, entry *filer_pb.Entry, ttl time.Duration) {
  188. if entry == nil {
  189. wfs.listDirectoryEntriesCache.Delete(string(path))
  190. } else {
  191. wfs.listDirectoryEntriesCache.Set(string(path), entry, ttl)
  192. }
  193. }
  194. func (wfs *WFS) cacheDelete(path util.FullPath) {
  195. wfs.listDirectoryEntriesCache.Delete(string(path))
  196. }
  197. func (wfs *WFS) AdjustedUrl(hostAndPort string) string {
  198. if !wfs.option.OutsideContainerClusterMode {
  199. return hostAndPort
  200. }
  201. commaIndex := strings.Index(hostAndPort, ":")
  202. if commaIndex < 0 {
  203. return hostAndPort
  204. }
  205. filerCommaIndex := strings.Index(wfs.option.FilerGrpcAddress, ":")
  206. return fmt.Sprintf("%s:%s", wfs.option.FilerGrpcAddress[:filerCommaIndex], hostAndPort[commaIndex+1:])
  207. }