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.

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