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.

315 lines
9.1 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
3 years ago
3 years ago
  1. package topology
  2. import (
  3. "encoding/json"
  4. "errors"
  5. "fmt"
  6. "github.com/chrislusf/seaweedfs/weed/pb"
  7. "github.com/chrislusf/seaweedfs/weed/storage/types"
  8. "math/rand"
  9. "sync"
  10. "time"
  11. "github.com/chrislusf/raft"
  12. hashicorpRaft "github.com/hashicorp/raft"
  13. "github.com/chrislusf/seaweedfs/weed/glog"
  14. "github.com/chrislusf/seaweedfs/weed/pb/master_pb"
  15. "github.com/chrislusf/seaweedfs/weed/sequence"
  16. "github.com/chrislusf/seaweedfs/weed/storage"
  17. "github.com/chrislusf/seaweedfs/weed/storage/needle"
  18. "github.com/chrislusf/seaweedfs/weed/storage/super_block"
  19. "github.com/chrislusf/seaweedfs/weed/util"
  20. )
  21. type Topology struct {
  22. vacuumLockCounter int64
  23. NodeImpl
  24. collectionMap *util.ConcurrentReadMap
  25. ecShardMap map[needle.VolumeId]*EcShardLocations
  26. ecShardMapLock sync.RWMutex
  27. pulse int64
  28. volumeSizeLimit uint64
  29. replicationAsMin bool
  30. Sequence sequence.Sequencer
  31. chanFullVolumes chan storage.VolumeInfo
  32. chanCrowdedVolumes chan storage.VolumeInfo
  33. Configuration *Configuration
  34. RaftServer raft.Server
  35. HashicorpRaft *hashicorpRaft.Raft
  36. }
  37. func NewTopology(id string, seq sequence.Sequencer, volumeSizeLimit uint64, pulse int, replicationAsMin bool) *Topology {
  38. t := &Topology{}
  39. t.id = NodeId(id)
  40. t.nodeType = "Topology"
  41. t.NodeImpl.value = t
  42. t.diskUsages = newDiskUsages()
  43. t.children = make(map[NodeId]Node)
  44. t.collectionMap = util.NewConcurrentReadMap()
  45. t.ecShardMap = make(map[needle.VolumeId]*EcShardLocations)
  46. t.pulse = int64(pulse)
  47. t.volumeSizeLimit = volumeSizeLimit
  48. t.replicationAsMin = replicationAsMin
  49. t.Sequence = seq
  50. t.chanFullVolumes = make(chan storage.VolumeInfo)
  51. t.chanCrowdedVolumes = make(chan storage.VolumeInfo)
  52. t.Configuration = &Configuration{}
  53. return t
  54. }
  55. func (t *Topology) IsLeader() bool {
  56. if t.RaftServer != nil {
  57. if t.RaftServer.State() == raft.Leader {
  58. return true
  59. }
  60. if leader, err := t.Leader(); err == nil {
  61. if pb.ServerAddress(t.RaftServer.Name()) == leader {
  62. return true
  63. }
  64. }
  65. } else if t.HashicorpRaft != nil {
  66. if t.HashicorpRaft.State() == hashicorpRaft.Leader {
  67. return true
  68. }
  69. }
  70. return false
  71. }
  72. func (t *Topology) Leader() (pb.ServerAddress, error) {
  73. var l pb.ServerAddress
  74. for count := 0; count < 3; count++ {
  75. if t.RaftServer != nil {
  76. l = pb.ServerAddress(t.RaftServer.Leader())
  77. } else if t.HashicorpRaft != nil {
  78. l = pb.ServerAddress(t.HashicorpRaft.Leader())
  79. } else {
  80. return "", errors.New("Raft Server not ready yet!")
  81. }
  82. if l != "" {
  83. break
  84. } else {
  85. time.Sleep(time.Duration(5+count) * time.Second)
  86. }
  87. }
  88. return l, nil
  89. }
  90. func (t *Topology) Lookup(collection string, vid needle.VolumeId) (dataNodes []*DataNode) {
  91. // maybe an issue if lots of collections?
  92. if collection == "" {
  93. for _, c := range t.collectionMap.Items() {
  94. if list := c.(*Collection).Lookup(vid); list != nil {
  95. return list
  96. }
  97. }
  98. } else {
  99. if c, ok := t.collectionMap.Find(collection); ok {
  100. return c.(*Collection).Lookup(vid)
  101. }
  102. }
  103. if locations, found := t.LookupEcShards(vid); found {
  104. for _, loc := range locations.Locations {
  105. dataNodes = append(dataNodes, loc...)
  106. }
  107. return dataNodes
  108. }
  109. return nil
  110. }
  111. func (t *Topology) NextVolumeId() (needle.VolumeId, error) {
  112. vid := t.GetMaxVolumeId()
  113. next := vid.Next()
  114. if t.RaftServer != nil {
  115. if _, err := t.RaftServer.Do(NewMaxVolumeIdCommand(next)); err != nil {
  116. return 0, err
  117. }
  118. } else if t.HashicorpRaft != nil {
  119. b, err := json.Marshal(NewMaxVolumeIdCommand(next))
  120. if err != nil {
  121. return 0, fmt.Errorf("failed marshal NewMaxVolumeIdCommand: %+v", err)
  122. }
  123. if future := t.HashicorpRaft.Apply(b, time.Second); future.Error() != nil {
  124. return 0, future.Error()
  125. }
  126. }
  127. return next, nil
  128. }
  129. // deprecated
  130. func (t *Topology) HasWritableVolume(option *VolumeGrowOption) bool {
  131. vl := t.GetVolumeLayout(option.Collection, option.ReplicaPlacement, option.Ttl, option.DiskType)
  132. active, _ := vl.GetActiveVolumeCount(option)
  133. return active > 0
  134. }
  135. func (t *Topology) PickForWrite(count uint64, option *VolumeGrowOption) (string, uint64, *VolumeLocationList, error) {
  136. vid, count, datanodes, err := t.GetVolumeLayout(option.Collection, option.ReplicaPlacement, option.Ttl, option.DiskType).PickForWrite(count, option)
  137. if err != nil {
  138. 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)
  139. }
  140. if datanodes.Length() == 0 {
  141. return "", 0, nil, fmt.Errorf("no writable volumes available for collection:%s replication:%s ttl:%s", option.Collection, option.ReplicaPlacement.String(), option.Ttl.String())
  142. }
  143. fileId := t.Sequence.NextFileId(count)
  144. return needle.NewFileId(*vid, fileId, rand.Uint32()).String(), count, datanodes, nil
  145. }
  146. func (t *Topology) GetVolumeLayout(collectionName string, rp *super_block.ReplicaPlacement, ttl *needle.TTL, diskType types.DiskType) *VolumeLayout {
  147. return t.collectionMap.Get(collectionName, func() interface{} {
  148. return NewCollection(collectionName, t.volumeSizeLimit, t.replicationAsMin)
  149. }).(*Collection).GetOrCreateVolumeLayout(rp, ttl, diskType)
  150. }
  151. func (t *Topology) ListCollections(includeNormalVolumes, includeEcVolumes bool) (ret []string) {
  152. mapOfCollections := make(map[string]bool)
  153. for _, c := range t.collectionMap.Items() {
  154. mapOfCollections[c.(*Collection).Name] = true
  155. }
  156. if includeEcVolumes {
  157. t.ecShardMapLock.RLock()
  158. for _, ecVolumeLocation := range t.ecShardMap {
  159. mapOfCollections[ecVolumeLocation.Collection] = true
  160. }
  161. t.ecShardMapLock.RUnlock()
  162. }
  163. for k := range mapOfCollections {
  164. ret = append(ret, k)
  165. }
  166. return ret
  167. }
  168. func (t *Topology) FindCollection(collectionName string) (*Collection, bool) {
  169. c, hasCollection := t.collectionMap.Find(collectionName)
  170. if !hasCollection {
  171. return nil, false
  172. }
  173. return c.(*Collection), hasCollection
  174. }
  175. func (t *Topology) DeleteCollection(collectionName string) {
  176. t.collectionMap.Delete(collectionName)
  177. }
  178. func (t *Topology) DeleteLayout(collectionName string, rp *super_block.ReplicaPlacement, ttl *needle.TTL, diskType types.DiskType) {
  179. collection, found := t.FindCollection(collectionName)
  180. if !found {
  181. return
  182. }
  183. collection.DeleteVolumeLayout(rp, ttl, diskType)
  184. if len(collection.storageType2VolumeLayout.Items()) == 0 {
  185. t.DeleteCollection(collectionName)
  186. }
  187. }
  188. func (t *Topology) RegisterVolumeLayout(v storage.VolumeInfo, dn *DataNode) {
  189. diskType := types.ToDiskType(v.DiskType)
  190. vl := t.GetVolumeLayout(v.Collection, v.ReplicaPlacement, v.Ttl, diskType)
  191. vl.RegisterVolume(&v, dn)
  192. vl.EnsureCorrectWritables(&v)
  193. }
  194. func (t *Topology) UnRegisterVolumeLayout(v storage.VolumeInfo, dn *DataNode) {
  195. glog.Infof("removing volume info: %+v from %v", v, dn.id)
  196. diskType := types.ToDiskType(v.DiskType)
  197. volumeLayout := t.GetVolumeLayout(v.Collection, v.ReplicaPlacement, v.Ttl, diskType)
  198. volumeLayout.UnRegisterVolume(&v, dn)
  199. if volumeLayout.isEmpty() {
  200. t.DeleteLayout(v.Collection, v.ReplicaPlacement, v.Ttl, diskType)
  201. }
  202. }
  203. func (t *Topology) GetOrCreateDataCenter(dcName string) *DataCenter {
  204. for _, c := range t.Children() {
  205. dc := c.(*DataCenter)
  206. if string(dc.Id()) == dcName {
  207. return dc
  208. }
  209. }
  210. dc := NewDataCenter(dcName)
  211. t.LinkChildNode(dc)
  212. return dc
  213. }
  214. func (t *Topology) SyncDataNodeRegistration(volumes []*master_pb.VolumeInformationMessage, dn *DataNode) (newVolumes, deletedVolumes []storage.VolumeInfo) {
  215. // convert into in memory struct storage.VolumeInfo
  216. var volumeInfos []storage.VolumeInfo
  217. for _, v := range volumes {
  218. if vi, err := storage.NewVolumeInfo(v); err == nil {
  219. volumeInfos = append(volumeInfos, vi)
  220. } else {
  221. glog.V(0).Infof("Fail to convert joined volume information: %v", err)
  222. }
  223. }
  224. // find out the delta volumes
  225. var changedVolumes []storage.VolumeInfo
  226. newVolumes, deletedVolumes, changedVolumes = dn.UpdateVolumes(volumeInfos)
  227. for _, v := range newVolumes {
  228. t.RegisterVolumeLayout(v, dn)
  229. }
  230. for _, v := range deletedVolumes {
  231. t.UnRegisterVolumeLayout(v, dn)
  232. }
  233. for _, v := range changedVolumes {
  234. diskType := types.ToDiskType(v.DiskType)
  235. vl := t.GetVolumeLayout(v.Collection, v.ReplicaPlacement, v.Ttl, diskType)
  236. vl.EnsureCorrectWritables(&v)
  237. }
  238. return
  239. }
  240. func (t *Topology) IncrementalSyncDataNodeRegistration(newVolumes, deletedVolumes []*master_pb.VolumeShortInformationMessage, dn *DataNode) {
  241. var newVis, oldVis []storage.VolumeInfo
  242. for _, v := range newVolumes {
  243. vi, err := storage.NewVolumeInfoFromShort(v)
  244. if err != nil {
  245. glog.V(0).Infof("NewVolumeInfoFromShort %v: %v", v, err)
  246. continue
  247. }
  248. newVis = append(newVis, vi)
  249. }
  250. for _, v := range deletedVolumes {
  251. vi, err := storage.NewVolumeInfoFromShort(v)
  252. if err != nil {
  253. glog.V(0).Infof("NewVolumeInfoFromShort %v: %v", v, err)
  254. continue
  255. }
  256. oldVis = append(oldVis, vi)
  257. }
  258. dn.DeltaUpdateVolumes(newVis, oldVis)
  259. for _, vi := range newVis {
  260. t.RegisterVolumeLayout(vi, dn)
  261. }
  262. for _, vi := range oldVis {
  263. t.UnRegisterVolumeLayout(vi, dn)
  264. }
  265. return
  266. }
  267. func (t *Topology) DataNodeRegistration(dcName, rackName string, dn *DataNode) {
  268. if dn.Parent() != nil {
  269. return
  270. }
  271. // registration to topo
  272. dc := t.GetOrCreateDataCenter(dcName)
  273. rack := dc.GetOrCreateRack(rackName)
  274. rack.LinkChildNode(dn)
  275. glog.Infof("[%s] reLink To topo ", dn.Id())
  276. }