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.

194 lines
5.4 KiB

10 years ago
10 years ago
  1. package topology
  2. import (
  3. "errors"
  4. "io/ioutil"
  5. "math/rand"
  6. "github.com/chrislusf/weed-fs/go/glog"
  7. "github.com/chrislusf/weed-fs/go/operation"
  8. "github.com/chrislusf/weed-fs/go/sequence"
  9. "github.com/chrislusf/weed-fs/go/storage"
  10. "github.com/chrislusf/weed-fs/go/util"
  11. "github.com/goraft/raft"
  12. )
  13. type Topology struct {
  14. NodeImpl
  15. collectionMap *util.ConcurrentReadMap
  16. pulse int64
  17. volumeSizeLimit uint64
  18. Sequence sequence.Sequencer
  19. chanDeadDataNodes chan *DataNode
  20. chanRecoveredDataNodes chan *DataNode
  21. chanFullVolumes chan storage.VolumeInfo
  22. configuration *Configuration
  23. RaftServer raft.Server
  24. }
  25. func NewTopology(id string, confFile string, seq sequence.Sequencer, volumeSizeLimit uint64, pulse int) (*Topology, error) {
  26. t := &Topology{}
  27. t.id = NodeId(id)
  28. t.nodeType = "Topology"
  29. t.NodeImpl.value = t
  30. t.children = make(map[NodeId]Node)
  31. t.collectionMap = util.NewConcurrentReadMap()
  32. t.pulse = int64(pulse)
  33. t.volumeSizeLimit = volumeSizeLimit
  34. t.Sequence = seq
  35. t.chanDeadDataNodes = make(chan *DataNode)
  36. t.chanRecoveredDataNodes = make(chan *DataNode)
  37. t.chanFullVolumes = make(chan storage.VolumeInfo)
  38. err := t.loadConfiguration(confFile)
  39. return t, err
  40. }
  41. func (t *Topology) IsLeader() bool {
  42. if leader, e := t.Leader(); e == nil {
  43. return leader == t.RaftServer.Name()
  44. }
  45. return false
  46. }
  47. func (t *Topology) Leader() (string, error) {
  48. l := ""
  49. if t.RaftServer != nil {
  50. l = t.RaftServer.Leader()
  51. } else {
  52. return "", errors.New("Raft Server not ready yet!")
  53. }
  54. if l == "" {
  55. // We are a single node cluster, we are the leader
  56. return t.RaftServer.Name(), errors.New("Raft Server not initialized!")
  57. }
  58. return l, nil
  59. }
  60. func (t *Topology) loadConfiguration(configurationFile string) error {
  61. b, e := ioutil.ReadFile(configurationFile)
  62. if e == nil {
  63. t.configuration, e = NewConfiguration(b)
  64. return e
  65. } else {
  66. glog.V(0).Infoln("Using default configurations.")
  67. }
  68. return nil
  69. }
  70. func (t *Topology) Lookup(collection string, vid storage.VolumeId) []*DataNode {
  71. //maybe an issue if lots of collections?
  72. if collection == "" {
  73. for _, c := range t.collectionMap.Items {
  74. if list := c.(*Collection).Lookup(vid); list != nil {
  75. return list
  76. }
  77. }
  78. } else {
  79. if c, ok := t.collectionMap.Items[collection]; ok {
  80. return c.(*Collection).Lookup(vid)
  81. }
  82. }
  83. return nil
  84. }
  85. func (t *Topology) NextVolumeId() storage.VolumeId {
  86. vid := t.GetMaxVolumeId()
  87. next := vid.Next()
  88. go t.RaftServer.Do(NewMaxVolumeIdCommand(next))
  89. return next
  90. }
  91. func (t *Topology) HasWritableVolume(option *VolumeGrowOption) bool {
  92. vl := t.GetVolumeLayout(option.Collection, option.ReplicaPlacement, option.Ttl)
  93. return vl.GetActiveVolumeCount(option) > 0
  94. }
  95. func (t *Topology) PickForWrite(count int, option *VolumeGrowOption) (string, int, *DataNode, error) {
  96. vid, count, datanodes, err := t.GetVolumeLayout(option.Collection, option.ReplicaPlacement, option.Ttl).PickForWrite(count, option)
  97. if err != nil || datanodes.Length() == 0 {
  98. return "", 0, nil, errors.New("No writable volumes available!")
  99. }
  100. fileId, count := t.Sequence.NextFileId(count)
  101. return storage.NewFileId(*vid, fileId, rand.Uint32()).String(), count, datanodes.Head(), nil
  102. }
  103. func (t *Topology) GetVolumeLayout(collectionName string, rp *storage.ReplicaPlacement, ttl *storage.TTL) *VolumeLayout {
  104. return t.collectionMap.Get(collectionName, func() interface{} {
  105. return NewCollection(collectionName, t.volumeSizeLimit)
  106. }).(*Collection).GetOrCreateVolumeLayout(rp, ttl)
  107. }
  108. func (t *Topology) GetCollection(collectionName string) (*Collection, bool) {
  109. c, hasCollection := t.collectionMap.Items[collectionName]
  110. return c.(*Collection), hasCollection
  111. }
  112. func (t *Topology) DeleteCollection(collectionName string) {
  113. delete(t.collectionMap.Items, collectionName)
  114. }
  115. func (t *Topology) RegisterVolumeLayout(v storage.VolumeInfo, dn *DataNode) {
  116. t.GetVolumeLayout(v.Collection, v.ReplicaPlacement, v.Ttl).RegisterVolume(&v, dn)
  117. }
  118. func (t *Topology) UnRegisterVolumeLayout(v storage.VolumeInfo, dn *DataNode) {
  119. glog.Infof("removing volume info:%+v", v)
  120. t.GetVolumeLayout(v.Collection, v.ReplicaPlacement, v.Ttl).UnRegisterVolume(&v, dn)
  121. }
  122. func (t *Topology) ProcessJoinMessage(joinMessage *operation.JoinMessage) {
  123. t.Sequence.SetMax(*joinMessage.MaxFileKey)
  124. dcName, rackName := t.configuration.Locate(*joinMessage.Ip, *joinMessage.DataCenter, *joinMessage.Rack)
  125. dc := t.GetOrCreateDataCenter(dcName)
  126. rack := dc.GetOrCreateRack(rackName)
  127. dn := rack.FindDataNode(*joinMessage.Ip, int(*joinMessage.Port))
  128. if *joinMessage.IsInit && dn != nil {
  129. t.UnRegisterDataNode(dn)
  130. }
  131. adminPort := *joinMessage.Port
  132. if joinMessage.AdminPort != nil {
  133. adminPort = *joinMessage.AdminPort
  134. }
  135. dn = rack.GetOrCreateDataNode(*joinMessage.Ip,
  136. int(*joinMessage.Port), int(adminPort), *joinMessage.PublicUrl,
  137. int(*joinMessage.MaxVolumeCount))
  138. var volumeInfos []storage.VolumeInfo
  139. for _, v := range joinMessage.Volumes {
  140. if vi, err := storage.NewVolumeInfo(v); err == nil {
  141. volumeInfos = append(volumeInfos, vi)
  142. } else {
  143. glog.V(0).Infoln("Fail to convert joined volume information:", err.Error())
  144. }
  145. }
  146. deletedVolumes := dn.UpdateVolumes(volumeInfos)
  147. for _, v := range volumeInfos {
  148. t.RegisterVolumeLayout(v, dn)
  149. }
  150. for _, v := range deletedVolumes {
  151. t.UnRegisterVolumeLayout(v, dn)
  152. }
  153. }
  154. func (t *Topology) GetOrCreateDataCenter(dcName string) *DataCenter {
  155. for _, c := range t.Children() {
  156. dc := c.(*DataCenter)
  157. if string(dc.Id()) == dcName {
  158. return dc
  159. }
  160. }
  161. dc := NewDataCenter(dcName)
  162. t.LinkChildNode(dc)
  163. return dc
  164. }