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.

268 lines
6.9 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
4 years ago
5 years ago
4 years ago
5 years ago
4 years ago
4 years ago
4 years ago
4 years ago
4 years ago
4 years ago
4 years ago
4 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 filer_pb
  2. import (
  3. "context"
  4. "errors"
  5. "fmt"
  6. "io"
  7. "os"
  8. "strings"
  9. "time"
  10. "github.com/chrislusf/seaweedfs/weed/glog"
  11. "github.com/chrislusf/seaweedfs/weed/util"
  12. )
  13. var (
  14. OS_UID = uint32(os.Getuid())
  15. OS_GID = uint32(os.Getgid())
  16. )
  17. type FilerClient interface {
  18. WithFilerClient(fn func(SeaweedFilerClient) error) error
  19. }
  20. func GetEntry(filerClient FilerClient, fullFilePath util.FullPath) (entry *Entry, err error) {
  21. dir, name := fullFilePath.DirAndName()
  22. err = filerClient.WithFilerClient(func(client SeaweedFilerClient) error {
  23. request := &LookupDirectoryEntryRequest{
  24. Directory: dir,
  25. Name: name,
  26. }
  27. // glog.V(3).Infof("read %s request: %v", fullFilePath, request)
  28. resp, err := LookupEntry(client, request)
  29. if err != nil {
  30. if err == ErrNotFound {
  31. return nil
  32. }
  33. glog.V(3).Infof("read %s %v: %v", fullFilePath, resp, err)
  34. return err
  35. }
  36. if resp.Entry == nil {
  37. // glog.V(3).Infof("read %s entry: %v", fullFilePath, entry)
  38. return nil
  39. }
  40. entry = resp.Entry
  41. return nil
  42. })
  43. return
  44. }
  45. type EachEntryFunciton func(entry *Entry, isLast bool) error
  46. func ReadDirAllEntries(filerClient FilerClient, fullDirPath util.FullPath, prefix string, fn EachEntryFunciton) (err error) {
  47. var counter uint32
  48. var startFrom string
  49. var counterFunc = func(entry *Entry, isLast bool) error {
  50. counter++
  51. startFrom = entry.Name
  52. return fn(entry, isLast)
  53. }
  54. var paginationLimit uint32 = 10000
  55. if err = doList(filerClient, fullDirPath, prefix, counterFunc, "", false, paginationLimit); err != nil {
  56. return err
  57. }
  58. for counter == paginationLimit {
  59. counter = 0
  60. if err = doList(filerClient, fullDirPath, prefix, counterFunc, startFrom, false, paginationLimit); err != nil {
  61. return err
  62. }
  63. }
  64. return nil
  65. }
  66. func List(filerClient FilerClient, parentDirectoryPath, prefix string, fn EachEntryFunciton, startFrom string, inclusive bool, limit uint32) (err error) {
  67. return filerClient.WithFilerClient(func(client SeaweedFilerClient) error {
  68. return doSeaweedList(client, util.FullPath(parentDirectoryPath), prefix, fn, startFrom, inclusive, limit)
  69. })
  70. }
  71. func doList(filerClient FilerClient, fullDirPath util.FullPath, prefix string, fn EachEntryFunciton, startFrom string, inclusive bool, limit uint32) (err error) {
  72. return filerClient.WithFilerClient(func(client SeaweedFilerClient) error {
  73. return doSeaweedList(client, fullDirPath, prefix, fn, startFrom, inclusive, limit)
  74. })
  75. }
  76. func SeaweedList(client SeaweedFilerClient, parentDirectoryPath, prefix string, fn EachEntryFunciton, startFrom string, inclusive bool, limit uint32) (err error) {
  77. return doSeaweedList(client, util.FullPath(parentDirectoryPath), prefix, fn, startFrom, inclusive, limit)
  78. }
  79. func doSeaweedList(client SeaweedFilerClient, fullDirPath util.FullPath, prefix string, fn EachEntryFunciton, startFrom string, inclusive bool, limit uint32) (err error) {
  80. request := &ListEntriesRequest{
  81. Directory: string(fullDirPath),
  82. Prefix: prefix,
  83. StartFromFileName: startFrom,
  84. Limit: limit,
  85. InclusiveStartFrom: inclusive,
  86. }
  87. glog.V(4).Infof("read directory: %v", request)
  88. ctx, cancel := context.WithCancel(context.Background())
  89. defer cancel()
  90. stream, err := client.ListEntries(ctx, request)
  91. if err != nil {
  92. return fmt.Errorf("list %s: %v", fullDirPath, err)
  93. }
  94. var prevEntry *Entry
  95. for {
  96. resp, recvErr := stream.Recv()
  97. if recvErr != nil {
  98. if recvErr == io.EOF {
  99. if prevEntry != nil {
  100. if err := fn(prevEntry, true); err != nil {
  101. return err
  102. }
  103. }
  104. break
  105. } else {
  106. return recvErr
  107. }
  108. }
  109. if prevEntry != nil {
  110. if err := fn(prevEntry, false); err != nil {
  111. return err
  112. }
  113. }
  114. prevEntry = resp.Entry
  115. }
  116. return nil
  117. }
  118. func Exists(filerClient FilerClient, parentDirectoryPath string, entryName string, isDirectory bool) (exists bool, err error) {
  119. err = filerClient.WithFilerClient(func(client SeaweedFilerClient) error {
  120. request := &LookupDirectoryEntryRequest{
  121. Directory: parentDirectoryPath,
  122. Name: entryName,
  123. }
  124. glog.V(4).Infof("exists entry %v/%v: %v", parentDirectoryPath, entryName, request)
  125. resp, err := LookupEntry(client, request)
  126. if err != nil {
  127. if err == ErrNotFound {
  128. exists = false
  129. return nil
  130. }
  131. glog.V(0).Infof("exists entry %v: %v", request, err)
  132. return fmt.Errorf("exists entry %s/%s: %v", parentDirectoryPath, entryName, err)
  133. }
  134. exists = resp.Entry.IsDirectory == isDirectory
  135. return nil
  136. })
  137. return
  138. }
  139. func Mkdir(filerClient FilerClient, parentDirectoryPath string, dirName string, fn func(entry *Entry)) error {
  140. return filerClient.WithFilerClient(func(client SeaweedFilerClient) error {
  141. entry := &Entry{
  142. Name: dirName,
  143. IsDirectory: true,
  144. Attributes: &FuseAttributes{
  145. Mtime: time.Now().Unix(),
  146. Crtime: time.Now().Unix(),
  147. FileMode: uint32(0777 | os.ModeDir),
  148. Uid: OS_UID,
  149. Gid: OS_GID,
  150. },
  151. }
  152. if fn != nil {
  153. fn(entry)
  154. }
  155. request := &CreateEntryRequest{
  156. Directory: parentDirectoryPath,
  157. Entry: entry,
  158. }
  159. glog.V(1).Infof("mkdir: %v", request)
  160. if err := CreateEntry(client, request); err != nil {
  161. glog.V(0).Infof("mkdir %v: %v", request, err)
  162. return fmt.Errorf("mkdir %s/%s: %v", parentDirectoryPath, dirName, err)
  163. }
  164. return nil
  165. })
  166. }
  167. func MkFile(filerClient FilerClient, parentDirectoryPath string, fileName string, chunks []*FileChunk) error {
  168. return filerClient.WithFilerClient(func(client SeaweedFilerClient) error {
  169. entry := &Entry{
  170. Name: fileName,
  171. IsDirectory: false,
  172. Attributes: &FuseAttributes{
  173. Mtime: time.Now().Unix(),
  174. Crtime: time.Now().Unix(),
  175. FileMode: uint32(0770),
  176. Uid: OS_UID,
  177. Gid: OS_GID,
  178. },
  179. Chunks: chunks,
  180. }
  181. request := &CreateEntryRequest{
  182. Directory: parentDirectoryPath,
  183. Entry: entry,
  184. }
  185. glog.V(1).Infof("create file: %s/%s", parentDirectoryPath, fileName)
  186. if err := CreateEntry(client, request); err != nil {
  187. glog.V(0).Infof("create file %v:%v", request, err)
  188. return fmt.Errorf("create file %s/%s: %v", parentDirectoryPath, fileName, err)
  189. }
  190. return nil
  191. })
  192. }
  193. func Remove(filerClient FilerClient, parentDirectoryPath, name string, isDeleteData, isRecursive, ignoreRecursiveErr, isFromOtherCluster bool, signatures []int32) error {
  194. return filerClient.WithFilerClient(func(client SeaweedFilerClient) error {
  195. deleteEntryRequest := &DeleteEntryRequest{
  196. Directory: parentDirectoryPath,
  197. Name: name,
  198. IsDeleteData: isDeleteData,
  199. IsRecursive: isRecursive,
  200. IgnoreRecursiveError: ignoreRecursiveErr,
  201. IsFromOtherCluster: isFromOtherCluster,
  202. Signatures: signatures,
  203. }
  204. if resp, err := client.DeleteEntry(context.Background(), deleteEntryRequest); err != nil {
  205. if strings.Contains(err.Error(), ErrNotFound.Error()) {
  206. return nil
  207. }
  208. return err
  209. } else {
  210. if resp.Error != "" {
  211. if strings.Contains(resp.Error, ErrNotFound.Error()) {
  212. return nil
  213. }
  214. return errors.New(resp.Error)
  215. }
  216. }
  217. return nil
  218. })
  219. }