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.

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