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