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.

341 lines
9.9 KiB

7 years ago
7 years ago
7 years ago
7 years ago
7 years ago
5 years ago
5 years ago
5 years ago
6 years ago
5 years ago
6 years ago
  1. package weed_server
  2. import (
  3. "context"
  4. "fmt"
  5. "os"
  6. "path/filepath"
  7. "strconv"
  8. "strings"
  9. "time"
  10. "github.com/chrislusf/seaweedfs/weed/filer2"
  11. "github.com/chrislusf/seaweedfs/weed/glog"
  12. "github.com/chrislusf/seaweedfs/weed/operation"
  13. "github.com/chrislusf/seaweedfs/weed/pb/filer_pb"
  14. "github.com/chrislusf/seaweedfs/weed/pb/master_pb"
  15. )
  16. func (fs *FilerServer) LookupDirectoryEntry(ctx context.Context, req *filer_pb.LookupDirectoryEntryRequest) (*filer_pb.LookupDirectoryEntryResponse, error) {
  17. entry, err := fs.filer.FindEntry(ctx, filer2.FullPath(filepath.ToSlash(filepath.Join(req.Directory, req.Name))))
  18. if err == filer2.ErrNotFound {
  19. return &filer_pb.LookupDirectoryEntryResponse{}, nil
  20. }
  21. if err != nil {
  22. glog.V(3).Infof("LookupDirectoryEntry %s: %+v, ", filepath.Join(req.Directory, req.Name), err)
  23. return nil, err
  24. }
  25. return &filer_pb.LookupDirectoryEntryResponse{
  26. Entry: &filer_pb.Entry{
  27. Name: req.Name,
  28. IsDirectory: entry.IsDirectory(),
  29. Attributes: filer2.EntryAttributeToPb(entry),
  30. Chunks: entry.Chunks,
  31. Extended: entry.Extended,
  32. },
  33. }, nil
  34. }
  35. func (fs *FilerServer) ListEntries(req *filer_pb.ListEntriesRequest, stream filer_pb.SeaweedFiler_ListEntriesServer) error {
  36. limit := int(req.Limit)
  37. if limit == 0 {
  38. limit = fs.option.DirListingLimit
  39. }
  40. paginationLimit := filer2.PaginationSize
  41. if limit < paginationLimit {
  42. paginationLimit = limit
  43. }
  44. lastFileName := req.StartFromFileName
  45. includeLastFile := req.InclusiveStartFrom
  46. for limit > 0 {
  47. entries, err := fs.filer.ListDirectoryEntries(stream.Context(), filer2.FullPath(req.Directory), lastFileName, includeLastFile, paginationLimit)
  48. if err != nil {
  49. return err
  50. }
  51. if len(entries) == 0 {
  52. return nil
  53. }
  54. includeLastFile = false
  55. for _, entry := range entries {
  56. lastFileName = entry.Name()
  57. if req.Prefix != "" {
  58. if !strings.HasPrefix(entry.Name(), req.Prefix) {
  59. continue
  60. }
  61. }
  62. if err := stream.Send(&filer_pb.ListEntriesResponse{
  63. Entry: &filer_pb.Entry{
  64. Name: entry.Name(),
  65. IsDirectory: entry.IsDirectory(),
  66. Chunks: entry.Chunks,
  67. Attributes: filer2.EntryAttributeToPb(entry),
  68. Extended: entry.Extended,
  69. },
  70. }); err != nil {
  71. return err
  72. }
  73. limit--
  74. if limit == 0 {
  75. return nil
  76. }
  77. }
  78. if len(entries) < paginationLimit {
  79. break
  80. }
  81. }
  82. return nil
  83. }
  84. func (fs *FilerServer) LookupVolume(ctx context.Context, req *filer_pb.LookupVolumeRequest) (*filer_pb.LookupVolumeResponse, error) {
  85. resp := &filer_pb.LookupVolumeResponse{
  86. LocationsMap: make(map[string]*filer_pb.Locations),
  87. }
  88. for _, vidString := range req.VolumeIds {
  89. vid, err := strconv.Atoi(vidString)
  90. if err != nil {
  91. glog.V(1).Infof("Unknown volume id %d", vid)
  92. return nil, err
  93. }
  94. var locs []*filer_pb.Location
  95. locations, found := fs.filer.MasterClient.GetLocations(uint32(vid))
  96. if !found {
  97. continue
  98. }
  99. for _, loc := range locations {
  100. locs = append(locs, &filer_pb.Location{
  101. Url: loc.Url,
  102. PublicUrl: loc.PublicUrl,
  103. })
  104. }
  105. resp.LocationsMap[vidString] = &filer_pb.Locations{
  106. Locations: locs,
  107. }
  108. }
  109. return resp, nil
  110. }
  111. func (fs *FilerServer) CreateEntry(ctx context.Context, req *filer_pb.CreateEntryRequest) (resp *filer_pb.CreateEntryResponse, err error) {
  112. resp = &filer_pb.CreateEntryResponse{}
  113. fullpath := filer2.FullPath(filepath.ToSlash(filepath.Join(req.Directory, req.Entry.Name)))
  114. chunks, garbages := filer2.CompactFileChunks(req.Entry.Chunks)
  115. if req.Entry.Attributes == nil {
  116. glog.V(3).Infof("CreateEntry %s: nil attributes", filepath.Join(req.Directory, req.Entry.Name))
  117. resp.Error = fmt.Sprintf("can not create entry with empty attributes")
  118. return
  119. }
  120. createErr := fs.filer.CreateEntry(ctx, &filer2.Entry{
  121. FullPath: fullpath,
  122. Attr: filer2.PbToEntryAttribute(req.Entry.Attributes),
  123. Chunks: chunks,
  124. }, req.OExcl)
  125. if createErr == nil {
  126. fs.filer.DeleteChunks(garbages)
  127. } else {
  128. glog.V(3).Infof("CreateEntry %s: %v", filepath.Join(req.Directory, req.Entry.Name), createErr)
  129. resp.Error = createErr.Error()
  130. }
  131. return
  132. }
  133. func (fs *FilerServer) UpdateEntry(ctx context.Context, req *filer_pb.UpdateEntryRequest) (*filer_pb.UpdateEntryResponse, error) {
  134. fullpath := filepath.ToSlash(filepath.Join(req.Directory, req.Entry.Name))
  135. entry, err := fs.filer.FindEntry(ctx, filer2.FullPath(fullpath))
  136. if err != nil {
  137. return &filer_pb.UpdateEntryResponse{}, fmt.Errorf("not found %s: %v", fullpath, err)
  138. }
  139. // remove old chunks if not included in the new ones
  140. unusedChunks := filer2.MinusChunks(entry.Chunks, req.Entry.Chunks)
  141. chunks, garbages := filer2.CompactFileChunks(req.Entry.Chunks)
  142. newEntry := &filer2.Entry{
  143. FullPath: filer2.FullPath(filepath.ToSlash(filepath.Join(req.Directory, req.Entry.Name))),
  144. Attr: entry.Attr,
  145. Extended: req.Entry.Extended,
  146. Chunks: chunks,
  147. }
  148. glog.V(3).Infof("updating %s: %+v, chunks %d: %v => %+v, chunks %d: %v, extended: %v => %v",
  149. fullpath, entry.Attr, len(entry.Chunks), entry.Chunks,
  150. req.Entry.Attributes, len(req.Entry.Chunks), req.Entry.Chunks,
  151. entry.Extended, req.Entry.Extended)
  152. if req.Entry.Attributes != nil {
  153. if req.Entry.Attributes.Mtime != 0 {
  154. newEntry.Attr.Mtime = time.Unix(req.Entry.Attributes.Mtime, 0)
  155. }
  156. if req.Entry.Attributes.FileMode != 0 {
  157. newEntry.Attr.Mode = os.FileMode(req.Entry.Attributes.FileMode)
  158. }
  159. newEntry.Attr.Uid = req.Entry.Attributes.Uid
  160. newEntry.Attr.Gid = req.Entry.Attributes.Gid
  161. newEntry.Attr.Mime = req.Entry.Attributes.Mime
  162. newEntry.Attr.UserName = req.Entry.Attributes.UserName
  163. newEntry.Attr.GroupNames = req.Entry.Attributes.GroupName
  164. }
  165. if filer2.EqualEntry(entry, newEntry) {
  166. return &filer_pb.UpdateEntryResponse{}, err
  167. }
  168. if err = fs.filer.UpdateEntry(ctx, entry, newEntry); err == nil {
  169. fs.filer.DeleteChunks(unusedChunks)
  170. fs.filer.DeleteChunks(garbages)
  171. } else {
  172. glog.V(3).Infof("UpdateEntry %s: %v", filepath.Join(req.Directory, req.Entry.Name), err)
  173. }
  174. fs.filer.NotifyUpdateEvent(entry, newEntry, true)
  175. return &filer_pb.UpdateEntryResponse{}, err
  176. }
  177. func (fs *FilerServer) DeleteEntry(ctx context.Context, req *filer_pb.DeleteEntryRequest) (resp *filer_pb.DeleteEntryResponse, err error) {
  178. err = fs.filer.DeleteEntryMetaAndData(ctx, filer2.FullPath(filepath.ToSlash(filepath.Join(req.Directory, req.Name))), req.IsRecursive, req.IgnoreRecursiveError, req.IsDeleteData)
  179. resp = &filer_pb.DeleteEntryResponse{}
  180. if err != nil {
  181. resp.Error = err.Error()
  182. }
  183. return resp, nil
  184. }
  185. func (fs *FilerServer) StreamDeleteEntries(stream filer_pb.SeaweedFiler_StreamDeleteEntriesServer) error {
  186. for {
  187. req, err := stream.Recv()
  188. if err != nil {
  189. return fmt.Errorf("receive delete entry request: %v", err)
  190. }
  191. fullpath := filer2.FullPath(filepath.ToSlash(filepath.Join(req.Directory, req.Name)))
  192. err = fs.filer.DeleteEntryMetaAndData(context.Background(), fullpath, req.IsRecursive, req.IgnoreRecursiveError, req.IsDeleteData)
  193. resp := &filer_pb.DeleteEntryResponse{}
  194. if err != nil {
  195. resp.Error = err.Error()
  196. }
  197. if err := stream.Send(resp); err != nil {
  198. return err
  199. }
  200. }
  201. return nil
  202. }
  203. func (fs *FilerServer) AssignVolume(ctx context.Context, req *filer_pb.AssignVolumeRequest) (resp *filer_pb.AssignVolumeResponse, err error) {
  204. ttlStr := ""
  205. if req.TtlSec > 0 {
  206. ttlStr = strconv.Itoa(int(req.TtlSec))
  207. }
  208. collection, replication := fs.detectCollection(req.ParentPath, req.Collection, req.Replication)
  209. var altRequest *operation.VolumeAssignRequest
  210. dataCenter := req.DataCenter
  211. if dataCenter == "" {
  212. dataCenter = fs.option.DataCenter
  213. }
  214. assignRequest := &operation.VolumeAssignRequest{
  215. Count: uint64(req.Count),
  216. Replication: replication,
  217. Collection: collection,
  218. Ttl: ttlStr,
  219. DataCenter: dataCenter,
  220. }
  221. if dataCenter != "" {
  222. altRequest = &operation.VolumeAssignRequest{
  223. Count: uint64(req.Count),
  224. Replication: replication,
  225. Collection: collection,
  226. Ttl: ttlStr,
  227. DataCenter: "",
  228. }
  229. }
  230. assignResult, err := operation.Assign(fs.filer.GetMaster(), fs.grpcDialOption, assignRequest, altRequest)
  231. if err != nil {
  232. glog.V(3).Infof("AssignVolume: %v", err)
  233. return &filer_pb.AssignVolumeResponse{Error: fmt.Sprintf("assign volume: %v", err)}, nil
  234. }
  235. if assignResult.Error != "" {
  236. glog.V(3).Infof("AssignVolume error: %v", assignResult.Error)
  237. return &filer_pb.AssignVolumeResponse{Error: fmt.Sprintf("assign volume result: %v", assignResult.Error)}, nil
  238. }
  239. return &filer_pb.AssignVolumeResponse{
  240. FileId: assignResult.Fid,
  241. Count: int32(assignResult.Count),
  242. Url: assignResult.Url,
  243. PublicUrl: assignResult.PublicUrl,
  244. Auth: string(assignResult.Auth),
  245. Collection: collection,
  246. Replication: replication,
  247. }, nil
  248. }
  249. func (fs *FilerServer) DeleteCollection(ctx context.Context, req *filer_pb.DeleteCollectionRequest) (resp *filer_pb.DeleteCollectionResponse, err error) {
  250. err = fs.filer.MasterClient.WithClient(func(client master_pb.SeaweedClient) error {
  251. _, err := client.CollectionDelete(context.Background(), &master_pb.CollectionDeleteRequest{
  252. Name: req.GetCollection(),
  253. })
  254. return err
  255. })
  256. return &filer_pb.DeleteCollectionResponse{}, err
  257. }
  258. func (fs *FilerServer) Statistics(ctx context.Context, req *filer_pb.StatisticsRequest) (resp *filer_pb.StatisticsResponse, err error) {
  259. input := &master_pb.StatisticsRequest{
  260. Replication: req.Replication,
  261. Collection: req.Collection,
  262. Ttl: req.Ttl,
  263. }
  264. output, err := operation.Statistics(fs.filer.GetMaster(), fs.grpcDialOption, input)
  265. if err != nil {
  266. return nil, err
  267. }
  268. return &filer_pb.StatisticsResponse{
  269. TotalSize: output.TotalSize,
  270. UsedSize: output.UsedSize,
  271. FileCount: output.FileCount,
  272. }, nil
  273. }
  274. func (fs *FilerServer) GetFilerConfiguration(ctx context.Context, req *filer_pb.GetFilerConfigurationRequest) (resp *filer_pb.GetFilerConfigurationResponse, err error) {
  275. return &filer_pb.GetFilerConfigurationResponse{
  276. Masters: fs.option.Masters,
  277. Collection: fs.option.Collection,
  278. Replication: fs.option.DefaultReplication,
  279. MaxMb: uint32(fs.option.MaxMB),
  280. DirBuckets: fs.option.DirBucketsPath,
  281. }, nil
  282. }