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.

525 lines
16 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
4 years ago
4 years ago
4 years ago
4 years ago
5 years ago
5 years ago
5 years ago
4 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
6 years ago
5 years ago
4 years ago
5 years ago
  1. package weed_server
  2. import (
  3. "context"
  4. "fmt"
  5. "os"
  6. "path/filepath"
  7. "strconv"
  8. "time"
  9. "github.com/chrislusf/seaweedfs/weed/filer"
  10. "github.com/chrislusf/seaweedfs/weed/glog"
  11. "github.com/chrislusf/seaweedfs/weed/operation"
  12. "github.com/chrislusf/seaweedfs/weed/pb/filer_pb"
  13. "github.com/chrislusf/seaweedfs/weed/pb/master_pb"
  14. "github.com/chrislusf/seaweedfs/weed/storage/needle"
  15. "github.com/chrislusf/seaweedfs/weed/util"
  16. )
  17. func (fs *FilerServer) LookupDirectoryEntry(ctx context.Context, req *filer_pb.LookupDirectoryEntryRequest) (*filer_pb.LookupDirectoryEntryResponse, error) {
  18. glog.V(4).Infof("LookupDirectoryEntry %s", filepath.Join(req.Directory, req.Name))
  19. entry, err := fs.filer.FindEntry(ctx, util.JoinPath(req.Directory, req.Name))
  20. if err == filer_pb.ErrNotFound {
  21. return &filer_pb.LookupDirectoryEntryResponse{}, err
  22. }
  23. if err != nil {
  24. glog.V(3).Infof("LookupDirectoryEntry %s: %+v, ", filepath.Join(req.Directory, req.Name), err)
  25. return nil, err
  26. }
  27. return &filer_pb.LookupDirectoryEntryResponse{
  28. Entry: &filer_pb.Entry{
  29. Name: req.Name,
  30. IsDirectory: entry.IsDirectory(),
  31. Attributes: filer.EntryAttributeToPb(entry),
  32. Chunks: entry.Chunks,
  33. Extended: entry.Extended,
  34. HardLinkId: entry.HardLinkId,
  35. HardLinkCounter: entry.HardLinkCounter,
  36. Content: entry.Content,
  37. },
  38. }, nil
  39. }
  40. func (fs *FilerServer) ListEntries(req *filer_pb.ListEntriesRequest, stream filer_pb.SeaweedFiler_ListEntriesServer) (err error) {
  41. glog.V(4).Infof("ListEntries %v", req)
  42. limit := int(req.Limit)
  43. if limit == 0 {
  44. limit = fs.option.DirListingLimit
  45. }
  46. paginationLimit := filer.PaginationSize
  47. if limit < paginationLimit {
  48. paginationLimit = limit
  49. }
  50. lastFileName := req.StartFromFileName
  51. includeLastFile := req.InclusiveStartFrom
  52. var listErr error
  53. for limit > 0 {
  54. var hasEntries bool
  55. lastFileName, listErr = fs.filer.StreamListDirectoryEntries(stream.Context(), util.FullPath(req.Directory), lastFileName, includeLastFile, int64(paginationLimit), req.Prefix, "", "", func(entry *filer.Entry) bool {
  56. hasEntries = true
  57. if err = stream.Send(&filer_pb.ListEntriesResponse{
  58. Entry: &filer_pb.Entry{
  59. Name: entry.Name(),
  60. IsDirectory: entry.IsDirectory(),
  61. Chunks: entry.Chunks,
  62. Attributes: filer.EntryAttributeToPb(entry),
  63. Extended: entry.Extended,
  64. HardLinkId: entry.HardLinkId,
  65. HardLinkCounter: entry.HardLinkCounter,
  66. Content: entry.Content,
  67. },
  68. }); err != nil {
  69. return false
  70. }
  71. limit--
  72. if limit == 0 {
  73. return false
  74. }
  75. return true
  76. })
  77. if listErr != nil {
  78. return listErr
  79. }
  80. if err != nil {
  81. return err
  82. }
  83. if !hasEntries {
  84. return nil
  85. }
  86. includeLastFile = false
  87. }
  88. return nil
  89. }
  90. func (fs *FilerServer) LookupVolume(ctx context.Context, req *filer_pb.LookupVolumeRequest) (*filer_pb.LookupVolumeResponse, error) {
  91. resp := &filer_pb.LookupVolumeResponse{
  92. LocationsMap: make(map[string]*filer_pb.Locations),
  93. }
  94. for _, vidString := range req.VolumeIds {
  95. vid, err := strconv.Atoi(vidString)
  96. if err != nil {
  97. glog.V(1).Infof("Unknown volume id %d", vid)
  98. return nil, err
  99. }
  100. var locs []*filer_pb.Location
  101. locations, found := fs.filer.MasterClient.GetLocations(uint32(vid))
  102. if !found {
  103. continue
  104. }
  105. for _, loc := range locations {
  106. locs = append(locs, &filer_pb.Location{
  107. Url: loc.Url,
  108. PublicUrl: loc.PublicUrl,
  109. })
  110. }
  111. resp.LocationsMap[vidString] = &filer_pb.Locations{
  112. Locations: locs,
  113. }
  114. }
  115. return resp, nil
  116. }
  117. func (fs *FilerServer) lookupFileId(fileId string) (targetUrls []string, err error) {
  118. fid, err := needle.ParseFileIdFromString(fileId)
  119. if err != nil {
  120. return nil, err
  121. }
  122. locations, found := fs.filer.MasterClient.GetLocations(uint32(fid.VolumeId))
  123. if !found || len(locations) == 0 {
  124. return nil, fmt.Errorf("not found volume %d in %s", fid.VolumeId, fileId)
  125. }
  126. for _, loc := range locations {
  127. targetUrls = append(targetUrls, fmt.Sprintf("http://%s/%s", loc.Url, fileId))
  128. }
  129. return
  130. }
  131. func (fs *FilerServer) CreateEntry(ctx context.Context, req *filer_pb.CreateEntryRequest) (resp *filer_pb.CreateEntryResponse, err error) {
  132. glog.V(4).Infof("CreateEntry %v/%v", req.Directory, req.Entry.Name)
  133. resp = &filer_pb.CreateEntryResponse{}
  134. chunks, garbage, err2 := fs.cleanupChunks(util.Join(req.Directory, req.Entry.Name), nil, req.Entry)
  135. if err2 != nil {
  136. return &filer_pb.CreateEntryResponse{}, fmt.Errorf("CreateEntry cleanupChunks %s %s: %v", req.Directory, req.Entry.Name, err2)
  137. }
  138. createErr := fs.filer.CreateEntry(ctx, &filer.Entry{
  139. FullPath: util.JoinPath(req.Directory, req.Entry.Name),
  140. Attr: filer.PbToEntryAttribute(req.Entry.Attributes),
  141. Chunks: chunks,
  142. Extended: req.Entry.Extended,
  143. HardLinkId: filer.HardLinkId(req.Entry.HardLinkId),
  144. HardLinkCounter: req.Entry.HardLinkCounter,
  145. Content: req.Entry.Content,
  146. }, req.OExcl, req.IsFromOtherCluster, req.Signatures)
  147. if createErr == nil {
  148. fs.filer.DeleteChunks(garbage)
  149. } else {
  150. glog.V(3).Infof("CreateEntry %s: %v", filepath.Join(req.Directory, req.Entry.Name), createErr)
  151. resp.Error = createErr.Error()
  152. }
  153. return
  154. }
  155. func (fs *FilerServer) UpdateEntry(ctx context.Context, req *filer_pb.UpdateEntryRequest) (*filer_pb.UpdateEntryResponse, error) {
  156. glog.V(4).Infof("UpdateEntry %v", req)
  157. fullpath := util.Join(req.Directory, req.Entry.Name)
  158. entry, err := fs.filer.FindEntry(ctx, util.FullPath(fullpath))
  159. if err != nil {
  160. return &filer_pb.UpdateEntryResponse{}, fmt.Errorf("not found %s: %v", fullpath, err)
  161. }
  162. chunks, garbage, err2 := fs.cleanupChunks(fullpath, entry, req.Entry)
  163. if err2 != nil {
  164. return &filer_pb.UpdateEntryResponse{}, fmt.Errorf("UpdateEntry cleanupChunks %s: %v", fullpath, err2)
  165. }
  166. newEntry := &filer.Entry{
  167. FullPath: util.JoinPath(req.Directory, req.Entry.Name),
  168. Attr: entry.Attr,
  169. Extended: req.Entry.Extended,
  170. Chunks: chunks,
  171. HardLinkId: filer.HardLinkId(req.Entry.HardLinkId),
  172. HardLinkCounter: req.Entry.HardLinkCounter,
  173. Content: req.Entry.Content,
  174. }
  175. glog.V(3).Infof("updating %s: %+v, chunks %d: %v => %+v, chunks %d: %v, extended: %v => %v",
  176. fullpath, entry.Attr, len(entry.Chunks), entry.Chunks,
  177. req.Entry.Attributes, len(req.Entry.Chunks), req.Entry.Chunks,
  178. entry.Extended, req.Entry.Extended)
  179. if req.Entry.Attributes != nil {
  180. if req.Entry.Attributes.Mtime != 0 {
  181. newEntry.Attr.Mtime = time.Unix(req.Entry.Attributes.Mtime, 0)
  182. }
  183. if req.Entry.Attributes.FileMode != 0 {
  184. newEntry.Attr.Mode = os.FileMode(req.Entry.Attributes.FileMode)
  185. }
  186. newEntry.Attr.Uid = req.Entry.Attributes.Uid
  187. newEntry.Attr.Gid = req.Entry.Attributes.Gid
  188. newEntry.Attr.Mime = req.Entry.Attributes.Mime
  189. newEntry.Attr.UserName = req.Entry.Attributes.UserName
  190. newEntry.Attr.GroupNames = req.Entry.Attributes.GroupName
  191. }
  192. if filer.EqualEntry(entry, newEntry) {
  193. return &filer_pb.UpdateEntryResponse{}, err
  194. }
  195. if err = fs.filer.UpdateEntry(ctx, entry, newEntry); err == nil {
  196. fs.filer.DeleteChunks(garbage)
  197. fs.filer.NotifyUpdateEvent(ctx, entry, newEntry, true, req.IsFromOtherCluster, req.Signatures)
  198. } else {
  199. glog.V(3).Infof("UpdateEntry %s: %v", filepath.Join(req.Directory, req.Entry.Name), err)
  200. }
  201. return &filer_pb.UpdateEntryResponse{}, err
  202. }
  203. func (fs *FilerServer) cleanupChunks(fullpath string, existingEntry *filer.Entry, newEntry *filer_pb.Entry) (chunks, garbage []*filer_pb.FileChunk, err error) {
  204. // remove old chunks if not included in the new ones
  205. if existingEntry != nil {
  206. garbage, err = filer.MinusChunks(fs.lookupFileId, existingEntry.Chunks, newEntry.Chunks)
  207. if err != nil {
  208. return newEntry.Chunks, nil, fmt.Errorf("MinusChunks: %v", err)
  209. }
  210. }
  211. // files with manifest chunks are usually large and append only, skip calculating covered chunks
  212. manifestChunks, nonManifestChunks := filer.SeparateManifestChunks(newEntry.Chunks)
  213. chunks, coveredChunks := filer.CompactFileChunks(fs.lookupFileId, nonManifestChunks)
  214. garbage = append(garbage, coveredChunks...)
  215. if newEntry.Attributes != nil {
  216. so, _ := fs.detectStorageOption(fullpath,
  217. newEntry.Attributes.Collection,
  218. newEntry.Attributes.Replication,
  219. newEntry.Attributes.TtlSec,
  220. newEntry.Attributes.DiskType,
  221. "",
  222. "",
  223. ) // ignore readonly error for capacity needed to manifestize
  224. chunks, err = filer.MaybeManifestize(fs.saveAsChunk(so), chunks)
  225. if err != nil {
  226. // not good, but should be ok
  227. glog.V(0).Infof("MaybeManifestize: %v", err)
  228. }
  229. }
  230. chunks = append(chunks, manifestChunks...)
  231. return
  232. }
  233. func (fs *FilerServer) AppendToEntry(ctx context.Context, req *filer_pb.AppendToEntryRequest) (*filer_pb.AppendToEntryResponse, error) {
  234. glog.V(4).Infof("AppendToEntry %v", req)
  235. fullpath := util.NewFullPath(req.Directory, req.EntryName)
  236. var offset int64 = 0
  237. entry, err := fs.filer.FindEntry(ctx, fullpath)
  238. if err == filer_pb.ErrNotFound {
  239. entry = &filer.Entry{
  240. FullPath: fullpath,
  241. Attr: filer.Attr{
  242. Crtime: time.Now(),
  243. Mtime: time.Now(),
  244. Mode: os.FileMode(0644),
  245. Uid: OS_UID,
  246. Gid: OS_GID,
  247. },
  248. }
  249. } else {
  250. offset = int64(filer.TotalSize(entry.Chunks))
  251. }
  252. for _, chunk := range req.Chunks {
  253. chunk.Offset = offset
  254. offset += int64(chunk.Size)
  255. }
  256. entry.Chunks = append(entry.Chunks, req.Chunks...)
  257. so, err := fs.detectStorageOption(string(fullpath), entry.Collection, entry.Replication, entry.TtlSec, entry.DiskType, "", "")
  258. if err != nil {
  259. glog.Warningf("detectStorageOption: %v", err)
  260. return &filer_pb.AppendToEntryResponse{}, err
  261. }
  262. entry.Chunks, err = filer.MaybeManifestize(fs.saveAsChunk(so), entry.Chunks)
  263. if err != nil {
  264. // not good, but should be ok
  265. glog.V(0).Infof("MaybeManifestize: %v", err)
  266. }
  267. err = fs.filer.CreateEntry(context.Background(), entry, false, false, nil)
  268. return &filer_pb.AppendToEntryResponse{}, err
  269. }
  270. func (fs *FilerServer) DeleteEntry(ctx context.Context, req *filer_pb.DeleteEntryRequest) (resp *filer_pb.DeleteEntryResponse, err error) {
  271. glog.V(4).Infof("DeleteEntry %v", req)
  272. err = fs.filer.DeleteEntryMetaAndData(ctx, util.JoinPath(req.Directory, req.Name), req.IsRecursive, req.IgnoreRecursiveError, req.IsDeleteData, req.IsFromOtherCluster, req.Signatures)
  273. resp = &filer_pb.DeleteEntryResponse{}
  274. if err != nil && err != filer_pb.ErrNotFound {
  275. resp.Error = err.Error()
  276. }
  277. return resp, nil
  278. }
  279. func (fs *FilerServer) AssignVolume(ctx context.Context, req *filer_pb.AssignVolumeRequest) (resp *filer_pb.AssignVolumeResponse, err error) {
  280. so, err := fs.detectStorageOption(req.Path, req.Collection, req.Replication, req.TtlSec, req.DiskType, req.DataCenter, req.Rack)
  281. if err != nil {
  282. glog.V(3).Infof("AssignVolume: %v", err)
  283. return &filer_pb.AssignVolumeResponse{Error: fmt.Sprintf("assign volume: %v", err)}, nil
  284. }
  285. assignRequest, altRequest := so.ToAssignRequests(int(req.Count))
  286. assignResult, err := operation.Assign(fs.filer.GetMaster, fs.grpcDialOption, assignRequest, altRequest)
  287. if err != nil {
  288. glog.V(3).Infof("AssignVolume: %v", err)
  289. return &filer_pb.AssignVolumeResponse{Error: fmt.Sprintf("assign volume: %v", err)}, nil
  290. }
  291. if assignResult.Error != "" {
  292. glog.V(3).Infof("AssignVolume error: %v", assignResult.Error)
  293. return &filer_pb.AssignVolumeResponse{Error: fmt.Sprintf("assign volume result: %v", assignResult.Error)}, nil
  294. }
  295. return &filer_pb.AssignVolumeResponse{
  296. FileId: assignResult.Fid,
  297. Count: int32(assignResult.Count),
  298. Url: assignResult.Url,
  299. PublicUrl: assignResult.PublicUrl,
  300. Auth: string(assignResult.Auth),
  301. Collection: so.Collection,
  302. Replication: so.Replication,
  303. }, nil
  304. }
  305. func (fs *FilerServer) CollectionList(ctx context.Context, req *filer_pb.CollectionListRequest) (resp *filer_pb.CollectionListResponse, err error) {
  306. glog.V(4).Infof("CollectionList %v", req)
  307. resp = &filer_pb.CollectionListResponse{}
  308. err = fs.filer.MasterClient.WithClient(func(client master_pb.SeaweedClient) error {
  309. masterResp, err := client.CollectionList(context.Background(), &master_pb.CollectionListRequest{
  310. IncludeNormalVolumes: req.IncludeNormalVolumes,
  311. IncludeEcVolumes: req.IncludeEcVolumes,
  312. })
  313. if err != nil {
  314. return err
  315. }
  316. for _, c := range masterResp.Collections {
  317. resp.Collections = append(resp.Collections, &filer_pb.Collection{Name: c.Name})
  318. }
  319. return nil
  320. })
  321. return
  322. }
  323. func (fs *FilerServer) DeleteCollection(ctx context.Context, req *filer_pb.DeleteCollectionRequest) (resp *filer_pb.DeleteCollectionResponse, err error) {
  324. glog.V(4).Infof("DeleteCollection %v", req)
  325. err = fs.filer.MasterClient.WithClient(func(client master_pb.SeaweedClient) error {
  326. _, err := client.CollectionDelete(context.Background(), &master_pb.CollectionDeleteRequest{
  327. Name: req.GetCollection(),
  328. })
  329. return err
  330. })
  331. return &filer_pb.DeleteCollectionResponse{}, err
  332. }
  333. func (fs *FilerServer) Statistics(ctx context.Context, req *filer_pb.StatisticsRequest) (resp *filer_pb.StatisticsResponse, err error) {
  334. var output *master_pb.StatisticsResponse
  335. err = fs.filer.MasterClient.WithClient(func(masterClient master_pb.SeaweedClient) error {
  336. grpcResponse, grpcErr := masterClient.Statistics(context.Background(), &master_pb.StatisticsRequest{
  337. Replication: req.Replication,
  338. Collection: req.Collection,
  339. Ttl: req.Ttl,
  340. DiskType: req.DiskType,
  341. })
  342. if grpcErr != nil {
  343. return grpcErr
  344. }
  345. output = grpcResponse
  346. return nil
  347. })
  348. if err != nil {
  349. return nil, err
  350. }
  351. return &filer_pb.StatisticsResponse{
  352. TotalSize: output.TotalSize,
  353. UsedSize: output.UsedSize,
  354. FileCount: output.FileCount,
  355. }, nil
  356. }
  357. func (fs *FilerServer) GetFilerConfiguration(ctx context.Context, req *filer_pb.GetFilerConfigurationRequest) (resp *filer_pb.GetFilerConfigurationResponse, err error) {
  358. t := &filer_pb.GetFilerConfigurationResponse{
  359. Masters: fs.option.Masters,
  360. Collection: fs.option.Collection,
  361. Replication: fs.option.DefaultReplication,
  362. MaxMb: uint32(fs.option.MaxMB),
  363. DirBuckets: fs.filer.DirBucketsPath,
  364. Cipher: fs.filer.Cipher,
  365. Signature: fs.filer.Signature,
  366. MetricsAddress: fs.metricsAddress,
  367. MetricsIntervalSec: int32(fs.metricsIntervalSec),
  368. Version: util.Version(),
  369. }
  370. glog.V(4).Infof("GetFilerConfiguration: %v", t)
  371. return t, nil
  372. }
  373. func (fs *FilerServer) KeepConnected(stream filer_pb.SeaweedFiler_KeepConnectedServer) error {
  374. req, err := stream.Recv()
  375. if err != nil {
  376. return err
  377. }
  378. clientName := fmt.Sprintf("%s:%d", req.Name, req.GrpcPort)
  379. m := make(map[string]bool)
  380. for _, tp := range req.Resources {
  381. m[tp] = true
  382. }
  383. fs.brokersLock.Lock()
  384. fs.brokers[clientName] = m
  385. glog.V(0).Infof("+ broker %v", clientName)
  386. fs.brokersLock.Unlock()
  387. defer func() {
  388. fs.brokersLock.Lock()
  389. delete(fs.brokers, clientName)
  390. glog.V(0).Infof("- broker %v: %v", clientName, err)
  391. fs.brokersLock.Unlock()
  392. }()
  393. for {
  394. if err := stream.Send(&filer_pb.KeepConnectedResponse{}); err != nil {
  395. glog.V(0).Infof("send broker %v: %+v", clientName, err)
  396. return err
  397. }
  398. // println("replied")
  399. if _, err := stream.Recv(); err != nil {
  400. glog.V(0).Infof("recv broker %v: %v", clientName, err)
  401. return err
  402. }
  403. // println("received")
  404. }
  405. }
  406. func (fs *FilerServer) LocateBroker(ctx context.Context, req *filer_pb.LocateBrokerRequest) (resp *filer_pb.LocateBrokerResponse, err error) {
  407. resp = &filer_pb.LocateBrokerResponse{}
  408. fs.brokersLock.Lock()
  409. defer fs.brokersLock.Unlock()
  410. var localBrokers []*filer_pb.LocateBrokerResponse_Resource
  411. for b, m := range fs.brokers {
  412. if _, found := m[req.Resource]; found {
  413. resp.Found = true
  414. resp.Resources = []*filer_pb.LocateBrokerResponse_Resource{
  415. {
  416. GrpcAddresses: b,
  417. ResourceCount: int32(len(m)),
  418. },
  419. }
  420. return
  421. }
  422. localBrokers = append(localBrokers, &filer_pb.LocateBrokerResponse_Resource{
  423. GrpcAddresses: b,
  424. ResourceCount: int32(len(m)),
  425. })
  426. }
  427. resp.Resources = localBrokers
  428. return resp, nil
  429. }