filer_conf.go 5.0 KB

123456789101112131415161718192021222324252627282930313233343536373839404142434445464748495051525354555657585960616263646566676869707172737475767778798081828384858687888990919293949596979899100101102103104105106107108109110111112113114115116117118119120121122123124125126127128129130131132133134135136137138139140141142143144145146147148149150151152153154155156157158159160161162163164165166167168169170171172173174175176177178179180181182183184185186
  1. package filer
  2. import (
  3. "bytes"
  4. "context"
  5. "fmt"
  6. "github.com/seaweedfs/seaweedfs/weed/pb"
  7. "github.com/seaweedfs/seaweedfs/weed/wdclient"
  8. "google.golang.org/grpc"
  9. "io"
  10. "github.com/seaweedfs/seaweedfs/weed/glog"
  11. "github.com/seaweedfs/seaweedfs/weed/pb/filer_pb"
  12. "github.com/seaweedfs/seaweedfs/weed/util"
  13. "github.com/viant/ptrie"
  14. jsonpb "google.golang.org/protobuf/encoding/protojson"
  15. )
  16. const (
  17. DirectoryEtcRoot = "/etc/"
  18. DirectoryEtcSeaweedFS = "/etc/seaweedfs"
  19. DirectoryEtcRemote = "/etc/remote"
  20. FilerConfName = "filer.conf"
  21. IamConfigDirectory = "/etc/iam"
  22. IamIdentityFile = "identity.json"
  23. IamPoliciesFile = "policies.json"
  24. )
  25. type FilerConf struct {
  26. rules ptrie.Trie
  27. }
  28. func ReadFilerConf(filerGrpcAddress pb.ServerAddress, grpcDialOption grpc.DialOption, masterClient *wdclient.MasterClient) (*FilerConf, error) {
  29. var buf bytes.Buffer
  30. if err := pb.WithGrpcFilerClient(false, 0, filerGrpcAddress, grpcDialOption, func(client filer_pb.SeaweedFilerClient) error {
  31. if masterClient != nil {
  32. return ReadEntry(masterClient, client, DirectoryEtcSeaweedFS, FilerConfName, &buf)
  33. } else {
  34. content, err := ReadInsideFiler(client, DirectoryEtcSeaweedFS, FilerConfName)
  35. buf = *bytes.NewBuffer(content)
  36. return err
  37. }
  38. }); err != nil && err != filer_pb.ErrNotFound {
  39. return nil, fmt.Errorf("read %s/%s: %v", DirectoryEtcSeaweedFS, FilerConfName, err)
  40. }
  41. fc := NewFilerConf()
  42. if buf.Len() > 0 {
  43. if err := fc.LoadFromBytes(buf.Bytes()); err != nil {
  44. return nil, fmt.Errorf("parse %s/%s: %v", DirectoryEtcSeaweedFS, FilerConfName, err)
  45. }
  46. }
  47. return fc, nil
  48. }
  49. func NewFilerConf() (fc *FilerConf) {
  50. fc = &FilerConf{
  51. rules: ptrie.New(),
  52. }
  53. return fc
  54. }
  55. func (fc *FilerConf) loadFromFiler(filer *Filer) (err error) {
  56. filerConfPath := util.NewFullPath(DirectoryEtcSeaweedFS, FilerConfName)
  57. entry, err := filer.FindEntry(context.Background(), filerConfPath)
  58. if err != nil {
  59. if err == filer_pb.ErrNotFound {
  60. return nil
  61. }
  62. glog.Errorf("read filer conf entry %s: %v", filerConfPath, err)
  63. return
  64. }
  65. if len(entry.Content) > 0 {
  66. return fc.LoadFromBytes(entry.Content)
  67. }
  68. return fc.loadFromChunks(filer, entry.Content, entry.GetChunks(), entry.Size())
  69. }
  70. func (fc *FilerConf) loadFromChunks(filer *Filer, content []byte, chunks []*filer_pb.FileChunk, size uint64) (err error) {
  71. if len(content) == 0 {
  72. content, err = filer.readEntry(chunks, size)
  73. if err != nil {
  74. glog.Errorf("read filer conf content: %v", err)
  75. return
  76. }
  77. }
  78. return fc.LoadFromBytes(content)
  79. }
  80. func (fc *FilerConf) LoadFromBytes(data []byte) (err error) {
  81. conf := &filer_pb.FilerConf{}
  82. if err := jsonpb.Unmarshal(data, conf); err != nil {
  83. return err
  84. }
  85. return fc.doLoadConf(conf)
  86. }
  87. func (fc *FilerConf) doLoadConf(conf *filer_pb.FilerConf) (err error) {
  88. for _, location := range conf.Locations {
  89. err = fc.AddLocationConf(location)
  90. if err != nil {
  91. // this is not recoverable
  92. return nil
  93. }
  94. }
  95. return nil
  96. }
  97. func (fc *FilerConf) AddLocationConf(locConf *filer_pb.FilerConf_PathConf) (err error) {
  98. err = fc.rules.Put([]byte(locConf.LocationPrefix), locConf)
  99. if err != nil {
  100. glog.Errorf("put location prefix: %v", err)
  101. }
  102. return
  103. }
  104. func (fc *FilerConf) DeleteLocationConf(locationPrefix string) {
  105. rules := ptrie.New()
  106. fc.rules.Walk(func(key []byte, value interface{}) bool {
  107. if string(key) == locationPrefix {
  108. return true
  109. }
  110. key = bytes.Clone(key)
  111. _ = rules.Put(key, value)
  112. return true
  113. })
  114. fc.rules = rules
  115. return
  116. }
  117. func (fc *FilerConf) MatchStorageRule(path string) (pathConf *filer_pb.FilerConf_PathConf) {
  118. pathConf = &filer_pb.FilerConf_PathConf{}
  119. fc.rules.MatchPrefix([]byte(path), func(key []byte, value interface{}) bool {
  120. t := value.(*filer_pb.FilerConf_PathConf)
  121. mergePathConf(pathConf, t)
  122. return true
  123. })
  124. return pathConf
  125. }
  126. func (fc *FilerConf) GetCollectionTtls(collection string) (ttls map[string]string) {
  127. ttls = make(map[string]string)
  128. fc.rules.Walk(func(key []byte, value interface{}) bool {
  129. t := value.(*filer_pb.FilerConf_PathConf)
  130. if t.Collection == collection {
  131. ttls[t.LocationPrefix] = t.GetTtl()
  132. }
  133. return true
  134. })
  135. return ttls
  136. }
  137. // merge if values in b is not empty, merge them into a
  138. func mergePathConf(a, b *filer_pb.FilerConf_PathConf) {
  139. a.Collection = util.Nvl(b.Collection, a.Collection)
  140. a.Replication = util.Nvl(b.Replication, a.Replication)
  141. a.Ttl = util.Nvl(b.Ttl, a.Ttl)
  142. a.DiskType = util.Nvl(b.DiskType, a.DiskType)
  143. a.Fsync = b.Fsync || a.Fsync
  144. if b.VolumeGrowthCount > 0 {
  145. a.VolumeGrowthCount = b.VolumeGrowthCount
  146. }
  147. a.ReadOnly = b.ReadOnly || a.ReadOnly
  148. a.DataCenter = util.Nvl(b.DataCenter, a.DataCenter)
  149. a.Rack = util.Nvl(b.Rack, a.Rack)
  150. a.DataNode = util.Nvl(b.DataNode, a.DataNode)
  151. }
  152. func (fc *FilerConf) ToProto() *filer_pb.FilerConf {
  153. m := &filer_pb.FilerConf{}
  154. fc.rules.Walk(func(key []byte, value interface{}) bool {
  155. pathConf := value.(*filer_pb.FilerConf_PathConf)
  156. m.Locations = append(m.Locations, pathConf)
  157. return true
  158. })
  159. return m
  160. }
  161. func (fc *FilerConf) ToText(writer io.Writer) error {
  162. return ProtoToText(writer, fc.ToProto())
  163. }