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.

295 lines
8.4 KiB

6 years ago
6 years ago
6 years ago
10 years ago
3 years ago
3 years ago
4 years ago
7 years ago
6 years ago
7 years ago
6 years ago
7 years ago
adding locking to avoid nil VolumeLocationList fix panic: runtime error: invalid memory address or nil pointer dereference Oct 22 00:53:44 bedb-master1 weed[8055]: [signal SIGSEGV: segmentation violation code=0x1 addr=0x8 pc=0x17658da] Oct 22 00:53:44 bedb-master1 weed[8055]: goroutine 310 [running]: Oct 22 00:53:44 bedb-master1 weed[8055]: github.com/chrislusf/seaweedfs/weed/topology.(*VolumeLocationList).Length(...) Oct 22 00:53:44 bedb-master1 weed[8055]: #011/root/seaweedfs/weed/topology/volume_location_list.go:35 Oct 22 00:53:44 bedb-master1 weed[8055]: github.com/chrislusf/seaweedfs/weed/topology.(*VolumeLayout).enoughCopies(...) Oct 22 00:53:44 bedb-master1 weed[8055]: #011/root/seaweedfs/weed/topology/volume_layout.go:376 Oct 22 00:53:44 bedb-master1 weed[8055]: github.com/chrislusf/seaweedfs/weed/topology.(*VolumeLayout).ensureCorrectWritables(0xc000111d50, 0xc000b55438) Oct 22 00:53:44 bedb-master1 weed[8055]: #011/root/seaweedfs/weed/topology/volume_layout.go:202 +0x5a Oct 22 00:53:44 bedb-master1 weed[8055]: github.com/chrislusf/seaweedfs/weed/topology.(*Topology).SyncDataNodeRegistration(0xc00042ac60, 0xc001454d30, 0x1, 0x1, 0xc0005fc000, 0xc00135de40, 0x4, 0xc00135de50, 0x10, 0x10d, ...) Oct 22 00:53:44 bedb-master1 weed[8055]: #011/root/seaweedfs/weed/topology/topology.go:224 +0x616 Oct 22 00:53:44 bedb-master1 weed[8055]: github.com/chrislusf/seaweedfs/weed/server.(*MasterServer).SendHeartbeat(0xc000162700, 0x23b97c0, 0xc000ae2c90, 0x0, 0x0) Oct 22 00:53:44 bedb-master1 weed[8055]: #011/root/seaweedfs/weed/server/master_grpc_server.go:106 +0x325 Oct 22 00:53:44 bedb-master1 weed[8055]: github.com/chrislusf/seaweedfs/weed/pb/master_pb._Seaweed_SendHeartbeat_Handler(0x1f8e7c0, 0xc000162700, 0x23b0a60, 0xc00024b440, 0x3172c38, 0xc000ab7100) Oct 22 00:53:44 bedb-master1 weed[8055]: #011/root/seaweedfs/weed/pb/master_pb/master.pb.go:4250 +0xad Oct 22 00:53:44 bedb-master1 weed[8055]: google.golang.org/grpc.(*Server).processStreamingRPC(0xc0001f31e0, 0x23bb800, 0xc000ac5500, 0xc000ab7100, 0xc0001fea80, 0x311fec0, 0x0, 0x0, 0x0) Oct 22 00:53:44 bedb-master1 weed[8055]: #011/root/go/pkg/mod/google.golang.org/grpc@v1.29.1/server.go:1329 +0xcd8 Oct 22 00:53:44 bedb-master1 weed[8055]: google.golang.org/grpc.(*Server).handleStream(0xc0001f31e0, 0x23bb800, 0xc000ac5500, 0xc000ab7100, 0x0) Oct 22 00:53:44 bedb-master1 weed[8055]: #011/root/go/pkg/mod/google.golang.org/grpc@v1.29.1/server.go:1409 +0xc5c Oct 22 00:53:44 bedb-master1 weed[8055]: google.golang.org/grpc.(*Server).serveStreams.func1.1(0xc0001ce8b0, 0xc0001f31e0, 0x23bb800, 0xc000ac5500, 0xc000ab7100) Oct 22 00:53:44 bedb-master1 weed[8055]: #011/root/go/pkg/mod/google.golang.org/grpc@v1.29.1/server.go:746 +0xa5 Oct 22 00:53:44 bedb-master1 weed[8055]: created by google.golang.org/grpc.(*Server).serveStreams.func1 Oct 22 00:53:44 bedb-master1 weed[8055]: #011/root/go/pkg/mod/google.golang.org/grpc@v1.29.1/server.go:744 +0xa5 Oct 22 00:53:44 bedb-master1 systemd[1]: weedmaster.service: Main process exited, code=exited, status=2/INVALIDARGUMENT Oct 22 00:53:44 bedb-master1 systemd[1]: weedmaster.service: Failed with result 'exit-code'.
4 years ago
7 years ago
  1. package topology
  2. import (
  3. "errors"
  4. "fmt"
  5. "github.com/chrislusf/seaweedfs/weed/pb"
  6. "github.com/chrislusf/seaweedfs/weed/storage/types"
  7. "math/rand"
  8. "sync"
  9. "time"
  10. "github.com/chrislusf/raft"
  11. "github.com/chrislusf/seaweedfs/weed/glog"
  12. "github.com/chrislusf/seaweedfs/weed/pb/master_pb"
  13. "github.com/chrislusf/seaweedfs/weed/sequence"
  14. "github.com/chrislusf/seaweedfs/weed/storage"
  15. "github.com/chrislusf/seaweedfs/weed/storage/needle"
  16. "github.com/chrislusf/seaweedfs/weed/storage/super_block"
  17. "github.com/chrislusf/seaweedfs/weed/util"
  18. )
  19. type Topology struct {
  20. vacuumLockCounter int64
  21. NodeImpl
  22. collectionMap *util.ConcurrentReadMap
  23. ecShardMap map[needle.VolumeId]*EcShardLocations
  24. ecShardMapLock sync.RWMutex
  25. pulse int64
  26. volumeSizeLimit uint64
  27. replicationAsMin bool
  28. Sequence sequence.Sequencer
  29. chanFullVolumes chan storage.VolumeInfo
  30. chanCrowdedVolumes chan storage.VolumeInfo
  31. Configuration *Configuration
  32. RaftServer raft.Server
  33. }
  34. func NewTopology(id string, seq sequence.Sequencer, volumeSizeLimit uint64, pulse int, replicationAsMin bool) *Topology {
  35. t := &Topology{}
  36. t.id = NodeId(id)
  37. t.nodeType = "Topology"
  38. t.NodeImpl.value = t
  39. t.diskUsages = newDiskUsages()
  40. t.children = make(map[NodeId]Node)
  41. t.collectionMap = util.NewConcurrentReadMap()
  42. t.ecShardMap = make(map[needle.VolumeId]*EcShardLocations)
  43. t.pulse = int64(pulse)
  44. t.volumeSizeLimit = volumeSizeLimit
  45. t.replicationAsMin = replicationAsMin
  46. t.Sequence = seq
  47. t.chanFullVolumes = make(chan storage.VolumeInfo)
  48. t.chanCrowdedVolumes = make(chan storage.VolumeInfo)
  49. t.Configuration = &Configuration{}
  50. return t
  51. }
  52. func (t *Topology) IsLeader() bool {
  53. if t.RaftServer != nil {
  54. if t.RaftServer.State() == raft.Leader {
  55. return true
  56. }
  57. if leader, err := t.Leader(); err == nil {
  58. if pb.ServerAddress(t.RaftServer.Name()) == leader {
  59. return true
  60. }
  61. }
  62. }
  63. return false
  64. }
  65. func (t *Topology) Leader() (pb.ServerAddress, error) {
  66. var l pb.ServerAddress
  67. for count := 0; count < 3; count++ {
  68. if t.RaftServer != nil {
  69. l = pb.ServerAddress(t.RaftServer.Leader())
  70. } else {
  71. return "", errors.New("Raft Server not ready yet!")
  72. }
  73. if l != "" {
  74. break
  75. } else {
  76. time.Sleep(time.Duration(5+count) * time.Second)
  77. }
  78. }
  79. return l, nil
  80. }
  81. func (t *Topology) Lookup(collection string, vid needle.VolumeId) (dataNodes []*DataNode) {
  82. // maybe an issue if lots of collections?
  83. if collection == "" {
  84. for _, c := range t.collectionMap.Items() {
  85. if list := c.(*Collection).Lookup(vid); list != nil {
  86. return list
  87. }
  88. }
  89. } else {
  90. if c, ok := t.collectionMap.Find(collection); ok {
  91. return c.(*Collection).Lookup(vid)
  92. }
  93. }
  94. if locations, found := t.LookupEcShards(vid); found {
  95. for _, loc := range locations.Locations {
  96. dataNodes = append(dataNodes, loc...)
  97. }
  98. return dataNodes
  99. }
  100. return nil
  101. }
  102. func (t *Topology) NextVolumeId() (needle.VolumeId, error) {
  103. vid := t.GetMaxVolumeId()
  104. next := vid.Next()
  105. if _, err := t.RaftServer.Do(NewMaxVolumeIdCommand(next)); err != nil {
  106. return 0, err
  107. }
  108. return next, nil
  109. }
  110. // deprecated
  111. func (t *Topology) HasWritableVolume(option *VolumeGrowOption) bool {
  112. vl := t.GetVolumeLayout(option.Collection, option.ReplicaPlacement, option.Ttl, option.DiskType)
  113. active, _ := vl.GetActiveVolumeCount(option)
  114. return active > 0
  115. }
  116. func (t *Topology) PickForWrite(count uint64, option *VolumeGrowOption) (string, uint64, *VolumeLocationList, error) {
  117. vid, count, datanodes, err := t.GetVolumeLayout(option.Collection, option.ReplicaPlacement, option.Ttl, option.DiskType).PickForWrite(count, option)
  118. if err != nil {
  119. return "", 0, nil, fmt.Errorf("failed to find writable volumes for collection:%s replication:%s ttl:%s error: %v", option.Collection, option.ReplicaPlacement.String(), option.Ttl.String(), err)
  120. }
  121. if datanodes.Length() == 0 {
  122. return "", 0, nil, fmt.Errorf("no writable volumes available for collection:%s replication:%s ttl:%s", option.Collection, option.ReplicaPlacement.String(), option.Ttl.String())
  123. }
  124. fileId := t.Sequence.NextFileId(count)
  125. return needle.NewFileId(*vid, fileId, rand.Uint32()).String(), count, datanodes, nil
  126. }
  127. func (t *Topology) GetVolumeLayout(collectionName string, rp *super_block.ReplicaPlacement, ttl *needle.TTL, diskType types.DiskType) *VolumeLayout {
  128. return t.collectionMap.Get(collectionName, func() interface{} {
  129. return NewCollection(collectionName, t.volumeSizeLimit, t.replicationAsMin)
  130. }).(*Collection).GetOrCreateVolumeLayout(rp, ttl, diskType)
  131. }
  132. func (t *Topology) ListCollections(includeNormalVolumes, includeEcVolumes bool) (ret []string) {
  133. mapOfCollections := make(map[string]bool)
  134. for _, c := range t.collectionMap.Items() {
  135. mapOfCollections[c.(*Collection).Name] = true
  136. }
  137. if includeEcVolumes {
  138. t.ecShardMapLock.RLock()
  139. for _, ecVolumeLocation := range t.ecShardMap {
  140. mapOfCollections[ecVolumeLocation.Collection] = true
  141. }
  142. t.ecShardMapLock.RUnlock()
  143. }
  144. for k := range mapOfCollections {
  145. ret = append(ret, k)
  146. }
  147. return ret
  148. }
  149. func (t *Topology) FindCollection(collectionName string) (*Collection, bool) {
  150. c, hasCollection := t.collectionMap.Find(collectionName)
  151. if !hasCollection {
  152. return nil, false
  153. }
  154. return c.(*Collection), hasCollection
  155. }
  156. func (t *Topology) DeleteCollection(collectionName string) {
  157. t.collectionMap.Delete(collectionName)
  158. }
  159. func (t *Topology) DeleteLayout(collectionName string, rp *super_block.ReplicaPlacement, ttl *needle.TTL, diskType types.DiskType) {
  160. collection, found := t.FindCollection(collectionName)
  161. if !found {
  162. return
  163. }
  164. collection.DeleteVolumeLayout(rp, ttl, diskType)
  165. if len(collection.storageType2VolumeLayout.Items()) == 0 {
  166. t.DeleteCollection(collectionName)
  167. }
  168. }
  169. func (t *Topology) RegisterVolumeLayout(v storage.VolumeInfo, dn *DataNode) {
  170. diskType := types.ToDiskType(v.DiskType)
  171. vl := t.GetVolumeLayout(v.Collection, v.ReplicaPlacement, v.Ttl, diskType)
  172. vl.RegisterVolume(&v, dn)
  173. vl.EnsureCorrectWritables(&v)
  174. }
  175. func (t *Topology) UnRegisterVolumeLayout(v storage.VolumeInfo, dn *DataNode) {
  176. glog.Infof("removing volume info: %+v from %v", v, dn.id)
  177. diskType := types.ToDiskType(v.DiskType)
  178. volumeLayout := t.GetVolumeLayout(v.Collection, v.ReplicaPlacement, v.Ttl, diskType)
  179. volumeLayout.UnRegisterVolume(&v, dn)
  180. if volumeLayout.isEmpty() {
  181. t.DeleteLayout(v.Collection, v.ReplicaPlacement, v.Ttl, diskType)
  182. }
  183. }
  184. func (t *Topology) GetOrCreateDataCenter(dcName string) *DataCenter {
  185. for _, c := range t.Children() {
  186. dc := c.(*DataCenter)
  187. if string(dc.Id()) == dcName {
  188. return dc
  189. }
  190. }
  191. dc := NewDataCenter(dcName)
  192. t.LinkChildNode(dc)
  193. return dc
  194. }
  195. func (t *Topology) SyncDataNodeRegistration(volumes []*master_pb.VolumeInformationMessage, dn *DataNode) (newVolumes, deletedVolumes []storage.VolumeInfo) {
  196. // convert into in memory struct storage.VolumeInfo
  197. var volumeInfos []storage.VolumeInfo
  198. for _, v := range volumes {
  199. if vi, err := storage.NewVolumeInfo(v); err == nil {
  200. volumeInfos = append(volumeInfos, vi)
  201. } else {
  202. glog.V(0).Infof("Fail to convert joined volume information: %v", err)
  203. }
  204. }
  205. // find out the delta volumes
  206. var changedVolumes []storage.VolumeInfo
  207. newVolumes, deletedVolumes, changedVolumes = dn.UpdateVolumes(volumeInfos)
  208. for _, v := range newVolumes {
  209. t.RegisterVolumeLayout(v, dn)
  210. }
  211. for _, v := range deletedVolumes {
  212. t.UnRegisterVolumeLayout(v, dn)
  213. }
  214. for _, v := range changedVolumes {
  215. diskType := types.ToDiskType(v.DiskType)
  216. vl := t.GetVolumeLayout(v.Collection, v.ReplicaPlacement, v.Ttl, diskType)
  217. vl.EnsureCorrectWritables(&v)
  218. }
  219. return
  220. }
  221. func (t *Topology) IncrementalSyncDataNodeRegistration(newVolumes, deletedVolumes []*master_pb.VolumeShortInformationMessage, dn *DataNode) {
  222. var newVis, oldVis []storage.VolumeInfo
  223. for _, v := range newVolumes {
  224. vi, err := storage.NewVolumeInfoFromShort(v)
  225. if err != nil {
  226. glog.V(0).Infof("NewVolumeInfoFromShort %v: %v", v, err)
  227. continue
  228. }
  229. newVis = append(newVis, vi)
  230. }
  231. for _, v := range deletedVolumes {
  232. vi, err := storage.NewVolumeInfoFromShort(v)
  233. if err != nil {
  234. glog.V(0).Infof("NewVolumeInfoFromShort %v: %v", v, err)
  235. continue
  236. }
  237. oldVis = append(oldVis, vi)
  238. }
  239. dn.DeltaUpdateVolumes(newVis, oldVis)
  240. for _, vi := range newVis {
  241. t.RegisterVolumeLayout(vi, dn)
  242. }
  243. for _, vi := range oldVis {
  244. t.UnRegisterVolumeLayout(vi, dn)
  245. }
  246. return
  247. }
  248. func (t *Topology) DataNodeRegistration(dcName, rackName string ,dn *DataNode){
  249. if dn.Parent() != nil{
  250. return
  251. }
  252. // registration to topo
  253. dc := t.GetOrCreateDataCenter(dcName)
  254. rack := dc.GetOrCreateRack(rackName)
  255. rack.LinkChildNode(dn)
  256. glog.Infof("[%s] reLink To topo ", dn.Id())
  257. }