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.

98 lines
2.6 KiB

4 years ago
4 years ago
  1. package filer
  2. import (
  3. "bytes"
  4. "github.com/chrislusf/seaweedfs/weed/glog"
  5. "github.com/chrislusf/seaweedfs/weed/pb/filer_pb"
  6. "github.com/chrislusf/seaweedfs/weed/util"
  7. )
  8. // onMetadataChangeEvent is triggered after filer processed change events from local or remote filers
  9. func (f *Filer) onMetadataChangeEvent(event *filer_pb.SubscribeMetadataResponse) {
  10. f.maybeReloadFilerConfiguration(event)
  11. f.maybeReloadRemoteStorageConfigurationAndMapping(event)
  12. f.onBucketEvents(event)
  13. }
  14. func (f *Filer) onBucketEvents(event *filer_pb.SubscribeMetadataResponse) {
  15. message := event.EventNotification
  16. for _, sig := range message.Signatures {
  17. if sig == f.Signature {
  18. return
  19. }
  20. }
  21. if f.DirBucketsPath == event.Directory {
  22. if message.OldEntry == nil && message.NewEntry != nil {
  23. if message.NewEntry.IsDirectory {
  24. f.Store.OnBucketCreation(message.NewEntry.Name)
  25. }
  26. }
  27. if message.OldEntry != nil && message.NewEntry == nil {
  28. if message.OldEntry.IsDirectory {
  29. f.Store.OnBucketDeletion(message.OldEntry.Name)
  30. }
  31. }
  32. }
  33. }
  34. func (f *Filer) maybeReloadFilerConfiguration(event *filer_pb.SubscribeMetadataResponse) {
  35. if DirectoryEtcSeaweedFS != event.Directory {
  36. if DirectoryEtcSeaweedFS != event.EventNotification.NewParentPath {
  37. return
  38. }
  39. }
  40. entry := event.EventNotification.NewEntry
  41. if entry == nil {
  42. return
  43. }
  44. glog.V(0).Infof("procesing %v", event)
  45. if entry.Name == FilerConfName {
  46. f.reloadFilerConfiguration(entry)
  47. }
  48. }
  49. func (f *Filer) readEntry(chunks []*filer_pb.FileChunk, size uint64) ([]byte, error) {
  50. var buf bytes.Buffer
  51. err := StreamContent(f.MasterClient, &buf, chunks, 0, int64(size))
  52. if err != nil {
  53. return nil, err
  54. }
  55. return buf.Bytes(), nil
  56. }
  57. func (f *Filer) reloadFilerConfiguration(entry *filer_pb.Entry) {
  58. fc := NewFilerConf()
  59. err := fc.loadFromChunks(f, entry.Content, entry.Chunks, FileSize(entry))
  60. if err != nil {
  61. glog.Errorf("read filer conf chunks: %v", err)
  62. return
  63. }
  64. f.FilerConf = fc
  65. }
  66. func (f *Filer) LoadFilerConf() {
  67. fc := NewFilerConf()
  68. err := util.Retry("loadFilerConf", func() error {
  69. return fc.loadFromFiler(f)
  70. })
  71. if err != nil {
  72. glog.Errorf("read filer conf: %v", err)
  73. return
  74. }
  75. f.FilerConf = fc
  76. }
  77. ////////////////////////////////////
  78. // load and maintain remote storages
  79. ////////////////////////////////////
  80. func (f *Filer) LoadRemoteStorageConfAndMapping() {
  81. if err := f.RemoteStorage.LoadRemoteStorageConfigurationsAndMapping(f); err != nil {
  82. glog.Errorf("read remote conf and mapping: %v", err)
  83. return
  84. }
  85. }
  86. func (f *Filer) maybeReloadRemoteStorageConfigurationAndMapping(event *filer_pb.SubscribeMetadataResponse) {
  87. // FIXME add reloading
  88. }