filer_on_meta_event.go 2.5 KB

12345678910111213141516171819202122232425262728293031323334353637383940414243444546474849505152535455565758596061626364656667686970717273747576777879808182838485868788899091929394
  1. package filer
  2. import (
  3. "bytes"
  4. "github.com/seaweedfs/seaweedfs/weed/glog"
  5. "github.com/seaweedfs/seaweedfs/weed/pb/filer_pb"
  6. "github.com/seaweedfs/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. if f.DirBucketsPath == event.Directory {
  17. if filer_pb.IsCreate(event) {
  18. if message.NewEntry.IsDirectory {
  19. f.Store.OnBucketCreation(message.NewEntry.Name)
  20. }
  21. }
  22. if filer_pb.IsDelete(event) {
  23. if message.OldEntry.IsDirectory {
  24. f.Store.OnBucketDeletion(message.OldEntry.Name)
  25. }
  26. }
  27. }
  28. }
  29. func (f *Filer) maybeReloadFilerConfiguration(event *filer_pb.SubscribeMetadataResponse) {
  30. if DirectoryEtcSeaweedFS != event.Directory {
  31. if DirectoryEtcSeaweedFS != event.EventNotification.NewParentPath {
  32. return
  33. }
  34. }
  35. entry := event.EventNotification.NewEntry
  36. if entry == nil {
  37. return
  38. }
  39. glog.V(0).Infof("procesing %v", event)
  40. if entry.Name == FilerConfName {
  41. f.reloadFilerConfiguration(entry)
  42. }
  43. }
  44. func (f *Filer) readEntry(chunks []*filer_pb.FileChunk, size uint64) ([]byte, error) {
  45. var buf bytes.Buffer
  46. err := StreamContent(f.MasterClient, &buf, chunks, 0, int64(size))
  47. if err != nil {
  48. return nil, err
  49. }
  50. return buf.Bytes(), nil
  51. }
  52. func (f *Filer) reloadFilerConfiguration(entry *filer_pb.Entry) {
  53. fc := NewFilerConf()
  54. err := fc.loadFromChunks(f, entry.Content, entry.GetChunks(), FileSize(entry))
  55. if err != nil {
  56. glog.Errorf("read filer conf chunks: %v", err)
  57. return
  58. }
  59. f.FilerConf = fc
  60. }
  61. func (f *Filer) LoadFilerConf() {
  62. fc := NewFilerConf()
  63. err := util.Retry("loadFilerConf", func() error {
  64. return fc.loadFromFiler(f)
  65. })
  66. if err != nil {
  67. glog.Errorf("read filer conf: %v", err)
  68. return
  69. }
  70. f.FilerConf = fc
  71. }
  72. // //////////////////////////////////
  73. // load and maintain remote storages
  74. // //////////////////////////////////
  75. func (f *Filer) LoadRemoteStorageConfAndMapping() {
  76. if err := f.RemoteStorage.LoadRemoteStorageConfigurationsAndMapping(f); err != nil {
  77. glog.Errorf("read remote conf and mapping: %v", err)
  78. return
  79. }
  80. }
  81. func (f *Filer) maybeReloadRemoteStorageConfigurationAndMapping(event *filer_pb.SubscribeMetadataResponse) {
  82. // FIXME add reloading
  83. }