filer_util.go 4.4 KB

123456789101112131415161718192021222324252627282930313233343536373839404142434445464748495051525354555657585960616263646566676869707172737475767778798081828384858687888990919293949596979899100101102103104105106107108109110111112113114115116117118119120121122123124125126127128129130131132133134135136137138139140141142143144145146147148149150151152153154155156157158159160161162163164165166167168169170171172
  1. package s3api
  2. import (
  3. "context"
  4. "fmt"
  5. "io"
  6. "os"
  7. "strings"
  8. "time"
  9. "github.com/chrislusf/seaweedfs/weed/glog"
  10. "github.com/chrislusf/seaweedfs/weed/pb/filer_pb"
  11. )
  12. func (s3a *S3ApiServer) mkdir(ctx context.Context, parentDirectoryPath string, dirName string, fn func(entry *filer_pb.Entry)) error {
  13. return s3a.withFilerClient(ctx, func(client filer_pb.SeaweedFilerClient) error {
  14. entry := &filer_pb.Entry{
  15. Name: dirName,
  16. IsDirectory: true,
  17. Attributes: &filer_pb.FuseAttributes{
  18. Mtime: time.Now().Unix(),
  19. Crtime: time.Now().Unix(),
  20. FileMode: uint32(0777 | os.ModeDir),
  21. Uid: OS_UID,
  22. Gid: OS_GID,
  23. },
  24. }
  25. if fn != nil {
  26. fn(entry)
  27. }
  28. request := &filer_pb.CreateEntryRequest{
  29. Directory: parentDirectoryPath,
  30. Entry: entry,
  31. }
  32. glog.V(1).Infof("mkdir: %v", request)
  33. if err := filer_pb.CreateEntry(ctx, client, request); err != nil {
  34. glog.V(0).Infof("mkdir %v: %v", request, err)
  35. return fmt.Errorf("mkdir %s/%s: %v", parentDirectoryPath, dirName, err)
  36. }
  37. return nil
  38. })
  39. }
  40. func (s3a *S3ApiServer) mkFile(ctx context.Context, parentDirectoryPath string, fileName string, chunks []*filer_pb.FileChunk) error {
  41. return s3a.withFilerClient(ctx, func(client filer_pb.SeaweedFilerClient) error {
  42. entry := &filer_pb.Entry{
  43. Name: fileName,
  44. IsDirectory: false,
  45. Attributes: &filer_pb.FuseAttributes{
  46. Mtime: time.Now().Unix(),
  47. Crtime: time.Now().Unix(),
  48. FileMode: uint32(0770),
  49. Uid: OS_UID,
  50. Gid: OS_GID,
  51. },
  52. Chunks: chunks,
  53. }
  54. request := &filer_pb.CreateEntryRequest{
  55. Directory: parentDirectoryPath,
  56. Entry: entry,
  57. }
  58. glog.V(1).Infof("create file: %s/%s", parentDirectoryPath, fileName)
  59. if err := filer_pb.CreateEntry(ctx, client, request); err != nil {
  60. glog.V(0).Infof("create file %v:%v", request, err)
  61. return fmt.Errorf("create file %s/%s: %v", parentDirectoryPath, fileName, err)
  62. }
  63. return nil
  64. })
  65. }
  66. func (s3a *S3ApiServer) list(ctx context.Context, parentDirectoryPath, prefix, startFrom string, inclusive bool, limit int) (entries []*filer_pb.Entry, err error) {
  67. err = s3a.withFilerClient(ctx, func(client filer_pb.SeaweedFilerClient) error {
  68. request := &filer_pb.ListEntriesRequest{
  69. Directory: parentDirectoryPath,
  70. Prefix: prefix,
  71. StartFromFileName: startFrom,
  72. InclusiveStartFrom: inclusive,
  73. Limit: uint32(limit),
  74. }
  75. glog.V(4).Infof("read directory: %v", request)
  76. stream, err := client.ListEntries(ctx, request)
  77. if err != nil {
  78. glog.V(0).Infof("read directory %v: %v", request, err)
  79. return fmt.Errorf("list dir %v: %v", parentDirectoryPath, err)
  80. }
  81. for {
  82. resp, recvErr := stream.Recv()
  83. if recvErr != nil {
  84. if recvErr == io.EOF {
  85. break
  86. } else {
  87. return recvErr
  88. }
  89. }
  90. entries = append(entries, resp.Entry)
  91. }
  92. return nil
  93. })
  94. return
  95. }
  96. func (s3a *S3ApiServer) rm(ctx context.Context, parentDirectoryPath string, entryName string, isDirectory, isDeleteData, isRecursive bool) error {
  97. return s3a.withFilerClient(ctx, func(client filer_pb.SeaweedFilerClient) error {
  98. request := &filer_pb.DeleteEntryRequest{
  99. Directory: parentDirectoryPath,
  100. Name: entryName,
  101. IsDeleteData: isDeleteData,
  102. IsRecursive: isRecursive,
  103. }
  104. glog.V(1).Infof("delete entry %v/%v: %v", parentDirectoryPath, entryName, request)
  105. if _, err := client.DeleteEntry(ctx, request); err != nil {
  106. glog.V(0).Infof("delete entry %v: %v", request, err)
  107. return fmt.Errorf("delete entry %s/%s: %v", parentDirectoryPath, entryName, err)
  108. }
  109. return nil
  110. })
  111. }
  112. func (s3a *S3ApiServer) exists(ctx context.Context, parentDirectoryPath string, entryName string, isDirectory bool) (exists bool, err error) {
  113. err = s3a.withFilerClient(ctx, func(client filer_pb.SeaweedFilerClient) error {
  114. request := &filer_pb.LookupDirectoryEntryRequest{
  115. Directory: parentDirectoryPath,
  116. Name: entryName,
  117. }
  118. glog.V(4).Infof("exists entry %v/%v: %v", parentDirectoryPath, entryName, request)
  119. resp, err := client.LookupDirectoryEntry(ctx, request)
  120. if err != nil {
  121. glog.V(0).Infof("exists entry %v: %v", request, err)
  122. return fmt.Errorf("exists entry %s/%s: %v", parentDirectoryPath, entryName, err)
  123. }
  124. exists = resp.Entry.IsDirectory == isDirectory
  125. return nil
  126. })
  127. return
  128. }
  129. func objectKey(key *string) *string {
  130. if strings.HasPrefix(*key, "/") {
  131. t := (*key)[1:]
  132. return &t
  133. }
  134. return key
  135. }