2012-08-29 15:58:03 +08:00
|
|
|
package topology
|
|
|
|
|
|
|
|
import (
|
2015-01-08 15:54:50 +08:00
|
|
|
"fmt"
|
2019-03-18 11:27:08 +08:00
|
|
|
"github.com/chrislusf/seaweedfs/weed/pb/master_pb"
|
2019-04-19 12:43:36 +08:00
|
|
|
"github.com/chrislusf/seaweedfs/weed/storage/needle"
|
2021-02-16 18:47:02 +08:00
|
|
|
"github.com/chrislusf/seaweedfs/weed/storage/types"
|
|
|
|
"github.com/chrislusf/seaweedfs/weed/util"
|
|
|
|
"strconv"
|
2019-04-19 12:43:36 +08:00
|
|
|
|
2016-06-03 09:09:14 +08:00
|
|
|
"github.com/chrislusf/seaweedfs/weed/glog"
|
|
|
|
"github.com/chrislusf/seaweedfs/weed/storage"
|
2012-08-29 15:58:03 +08:00
|
|
|
)
|
|
|
|
|
2012-09-09 07:25:44 +08:00
|
|
|
type DataNode struct {
|
2012-09-03 05:33:48 +08:00
|
|
|
NodeImpl
|
2021-02-16 19:03:00 +08:00
|
|
|
Ip string
|
|
|
|
Port int
|
|
|
|
PublicUrl string
|
|
|
|
LastSeen int64 // unix time in seconds
|
2012-08-29 15:58:03 +08:00
|
|
|
}
|
2012-09-03 05:33:48 +08:00
|
|
|
|
2012-09-09 07:25:44 +08:00
|
|
|
func NewDataNode(id string) *DataNode {
|
2021-02-16 18:47:02 +08:00
|
|
|
dn := &DataNode{}
|
|
|
|
dn.id = NodeId(id)
|
|
|
|
dn.nodeType = "DataNode"
|
|
|
|
dn.diskUsages = newDiskUsages()
|
|
|
|
dn.children = make(map[NodeId]Node)
|
|
|
|
dn.NodeImpl.value = dn
|
|
|
|
return dn
|
2012-08-31 16:35:11 +08:00
|
|
|
}
|
2014-04-13 16:29:52 +08:00
|
|
|
|
2015-01-08 15:54:50 +08:00
|
|
|
func (dn *DataNode) String() string {
|
2016-05-20 14:32:56 +08:00
|
|
|
dn.RLock()
|
|
|
|
defer dn.RUnlock()
|
2021-02-16 18:47:02 +08:00
|
|
|
return fmt.Sprintf("Node:%s, Ip:%s, Port:%d, PublicUrl:%s", dn.NodeImpl.String(), dn.Ip, dn.Port, dn.PublicUrl)
|
2015-01-08 15:54:50 +08:00
|
|
|
}
|
|
|
|
|
2020-06-05 23:18:15 +08:00
|
|
|
func (dn *DataNode) AddOrUpdateVolume(v storage.VolumeInfo) (isNew, isChangedRO bool) {
|
2016-05-20 14:32:56 +08:00
|
|
|
dn.Lock()
|
|
|
|
defer dn.Unlock()
|
2020-09-21 14:07:55 +08:00
|
|
|
return dn.doAddOrUpdateVolume(v)
|
|
|
|
}
|
|
|
|
|
2021-02-16 18:47:02 +08:00
|
|
|
func (dn *DataNode) getOrCreateDisk(diskType string) *Disk {
|
|
|
|
c, found := dn.children[NodeId(diskType)]
|
|
|
|
if !found {
|
|
|
|
c = NewDisk(diskType)
|
2021-02-17 02:48:16 +08:00
|
|
|
dn.doLinkChildNode(c)
|
2012-09-17 08:31:15 +08:00
|
|
|
}
|
2021-02-16 18:47:02 +08:00
|
|
|
disk := c.(*Disk)
|
|
|
|
return disk
|
|
|
|
}
|
|
|
|
|
|
|
|
func (dn *DataNode) doAddOrUpdateVolume(v storage.VolumeInfo) (isNew, isChangedRO bool) {
|
|
|
|
disk := dn.getOrCreateDisk(v.DiskType)
|
|
|
|
return disk.AddOrUpdateVolume(v)
|
2012-09-10 15:18:07 +08:00
|
|
|
}
|
2014-04-13 16:29:52 +08:00
|
|
|
|
2021-02-16 18:47:02 +08:00
|
|
|
// UpdateVolumes detects new/deleted/changed volumes on a volume server
|
|
|
|
// used in master to notify master clients of these changes.
|
2020-06-05 23:18:15 +08:00
|
|
|
func (dn *DataNode) UpdateVolumes(actualVolumes []storage.VolumeInfo) (newVolumes, deletedVolumes, changeRO []storage.VolumeInfo) {
|
2020-09-21 14:07:55 +08:00
|
|
|
|
2019-04-19 12:43:36 +08:00
|
|
|
actualVolumeMap := make(map[needle.VolumeId]storage.VolumeInfo)
|
2014-03-11 02:43:54 +08:00
|
|
|
for _, v := range actualVolumes {
|
|
|
|
actualVolumeMap[v.Id] = v
|
|
|
|
}
|
2020-09-21 14:07:55 +08:00
|
|
|
|
2016-09-08 09:13:49 +08:00
|
|
|
dn.Lock()
|
2020-09-21 14:07:55 +08:00
|
|
|
defer dn.Unlock()
|
|
|
|
|
2021-02-16 18:47:02 +08:00
|
|
|
existingVolumes := dn.getVolumes()
|
|
|
|
|
|
|
|
for _, v := range existingVolumes {
|
|
|
|
vid := v.Id
|
2014-03-11 02:43:54 +08:00
|
|
|
if _, ok := actualVolumeMap[vid]; !ok {
|
|
|
|
glog.V(0).Infoln("Deleting volume id:", vid)
|
2021-02-16 18:47:02 +08:00
|
|
|
disk := dn.getOrCreateDisk(v.DiskType)
|
|
|
|
delete(disk.volumes, vid)
|
2014-09-21 03:38:59 +08:00
|
|
|
deletedVolumes = append(deletedVolumes, v)
|
2021-02-16 18:47:02 +08:00
|
|
|
|
|
|
|
deltaDiskUsages := newDiskUsages()
|
2021-02-16 19:03:00 +08:00
|
|
|
deltaDiskUsage := deltaDiskUsages.getOrCreateDisk(types.ToDiskType(v.DiskType))
|
2021-02-16 18:47:02 +08:00
|
|
|
deltaDiskUsage.volumeCount = -1
|
2019-12-04 13:36:42 +08:00
|
|
|
if v.IsRemote() {
|
2021-02-16 18:47:02 +08:00
|
|
|
deltaDiskUsage.remoteVolumeCount = -1
|
2019-12-04 13:36:42 +08:00
|
|
|
}
|
|
|
|
if !v.ReadOnly {
|
2021-02-16 18:47:02 +08:00
|
|
|
deltaDiskUsage.activeVolumeCount = -1
|
2019-12-04 13:36:42 +08:00
|
|
|
}
|
2021-02-16 18:47:02 +08:00
|
|
|
disk.UpAdjustDiskUsageDelta(deltaDiskUsages)
|
2014-03-11 02:43:54 +08:00
|
|
|
}
|
2016-09-08 09:13:49 +08:00
|
|
|
}
|
2014-03-11 02:43:54 +08:00
|
|
|
for _, v := range actualVolumes {
|
2020-09-21 14:07:55 +08:00
|
|
|
isNew, isChangedRO := dn.doAddOrUpdateVolume(v)
|
2018-07-28 14:09:55 +08:00
|
|
|
if isNew {
|
|
|
|
newVolumes = append(newVolumes, v)
|
|
|
|
}
|
2020-06-05 23:18:15 +08:00
|
|
|
if isChangedRO {
|
|
|
|
changeRO = append(changeRO, v)
|
|
|
|
}
|
2014-03-11 02:43:54 +08:00
|
|
|
}
|
2014-09-21 03:38:59 +08:00
|
|
|
return
|
2014-03-11 02:43:54 +08:00
|
|
|
}
|
2014-04-13 16:29:52 +08:00
|
|
|
|
2020-09-21 14:07:55 +08:00
|
|
|
func (dn *DataNode) DeltaUpdateVolumes(newVolumes, deletedVolumes []storage.VolumeInfo) {
|
2019-04-21 02:35:20 +08:00
|
|
|
dn.Lock()
|
2020-09-21 14:07:55 +08:00
|
|
|
defer dn.Unlock()
|
|
|
|
|
2019-04-21 02:35:20 +08:00
|
|
|
for _, v := range deletedVolumes {
|
2021-02-16 18:47:02 +08:00
|
|
|
disk := dn.getOrCreateDisk(v.DiskType)
|
|
|
|
delete(disk.volumes, v.Id)
|
|
|
|
|
|
|
|
deltaDiskUsages := newDiskUsages()
|
2021-02-16 19:03:00 +08:00
|
|
|
deltaDiskUsage := deltaDiskUsages.getOrCreateDisk(types.ToDiskType(v.DiskType))
|
2021-02-16 18:47:02 +08:00
|
|
|
deltaDiskUsage.volumeCount = -1
|
2019-12-04 13:36:42 +08:00
|
|
|
if v.IsRemote() {
|
2021-02-16 18:47:02 +08:00
|
|
|
deltaDiskUsage.remoteVolumeCount = -1
|
2019-12-04 13:36:42 +08:00
|
|
|
}
|
|
|
|
if !v.ReadOnly {
|
2021-02-16 18:47:02 +08:00
|
|
|
deltaDiskUsage.activeVolumeCount = -1
|
2019-12-04 13:36:42 +08:00
|
|
|
}
|
2021-02-16 18:47:02 +08:00
|
|
|
disk.UpAdjustDiskUsageDelta(deltaDiskUsages)
|
2019-04-21 02:35:20 +08:00
|
|
|
}
|
2020-09-21 14:07:55 +08:00
|
|
|
for _, v := range newVolumes {
|
|
|
|
dn.doAddOrUpdateVolume(v)
|
2019-04-21 02:35:20 +08:00
|
|
|
}
|
|
|
|
return
|
|
|
|
}
|
|
|
|
|
2021-02-16 19:55:24 +08:00
|
|
|
func (dn *DataNode) AdjustMaxVolumeCounts(maxVolumeCounts map[string]uint32) {
|
|
|
|
deltaDiskUsages := newDiskUsages()
|
|
|
|
for diskType, maxVolumeCount := range maxVolumeCounts {
|
|
|
|
if maxVolumeCount == 0 {
|
|
|
|
// the volume server may have set the max to zero
|
|
|
|
continue
|
|
|
|
}
|
|
|
|
dt := types.ToDiskType(diskType)
|
|
|
|
currentDiskUsage := dn.diskUsages.getOrCreateDisk(dt)
|
|
|
|
if currentDiskUsage.maxVolumeCount == int64(maxVolumeCount) {
|
|
|
|
continue
|
|
|
|
}
|
|
|
|
disk := dn.getOrCreateDisk(dt.String())
|
|
|
|
deltaDiskUsage := deltaDiskUsages.getOrCreateDisk(dt)
|
|
|
|
deltaDiskUsage.maxVolumeCount = int64(maxVolumeCount) - currentDiskUsage.maxVolumeCount
|
|
|
|
disk.UpAdjustDiskUsageDelta(deltaDiskUsages)
|
|
|
|
}
|
|
|
|
}
|
|
|
|
|
2016-05-20 14:32:56 +08:00
|
|
|
func (dn *DataNode) GetVolumes() (ret []storage.VolumeInfo) {
|
|
|
|
dn.RLock()
|
2021-02-16 18:47:02 +08:00
|
|
|
for _, c := range dn.children {
|
|
|
|
disk := c.(*Disk)
|
|
|
|
ret = append(ret, disk.GetVolumes()...)
|
2016-05-20 14:32:56 +08:00
|
|
|
}
|
|
|
|
dn.RUnlock()
|
|
|
|
return ret
|
|
|
|
}
|
|
|
|
|
2021-02-16 18:47:02 +08:00
|
|
|
func (dn *DataNode) GetVolumesById(id needle.VolumeId) (vInfo storage.VolumeInfo, err error) {
|
2016-08-09 20:12:39 +08:00
|
|
|
dn.RLock()
|
|
|
|
defer dn.RUnlock()
|
2021-02-16 18:47:02 +08:00
|
|
|
found := false
|
|
|
|
for _, c := range dn.children {
|
|
|
|
disk := c.(*Disk)
|
|
|
|
vInfo, found = disk.volumes[id]
|
|
|
|
if found {
|
|
|
|
break
|
|
|
|
}
|
|
|
|
}
|
|
|
|
if found {
|
2019-01-17 09:17:19 +08:00
|
|
|
return vInfo, nil
|
2016-08-09 20:12:39 +08:00
|
|
|
} else {
|
|
|
|
return storage.VolumeInfo{}, fmt.Errorf("volumeInfo not found")
|
|
|
|
}
|
|
|
|
}
|
|
|
|
|
2013-06-20 09:10:38 +08:00
|
|
|
func (dn *DataNode) GetDataCenter() *DataCenter {
|
2021-01-04 04:25:58 +08:00
|
|
|
rack := dn.Parent()
|
|
|
|
dcNode := rack.Parent()
|
|
|
|
dcValue := dcNode.GetValue()
|
|
|
|
return dcValue.(*DataCenter)
|
2013-06-20 09:10:38 +08:00
|
|
|
}
|
2014-04-13 16:29:52 +08:00
|
|
|
|
|
|
|
func (dn *DataNode) GetRack() *Rack {
|
|
|
|
return dn.Parent().(*NodeImpl).value.(*Rack)
|
|
|
|
}
|
|
|
|
|
2012-09-10 15:18:07 +08:00
|
|
|
func (dn *DataNode) GetTopology() *Topology {
|
2013-06-20 09:10:38 +08:00
|
|
|
p := dn.Parent()
|
2012-09-17 08:31:15 +08:00
|
|
|
for p.Parent() != nil {
|
|
|
|
p = p.Parent()
|
|
|
|
}
|
|
|
|
t := p.(*Topology)
|
|
|
|
return t
|
2012-08-29 15:58:03 +08:00
|
|
|
}
|
2014-04-13 16:29:52 +08:00
|
|
|
|
2012-09-14 16:17:13 +08:00
|
|
|
func (dn *DataNode) MatchLocation(ip string, port int) bool {
|
2012-09-17 08:31:15 +08:00
|
|
|
return dn.Ip == ip && dn.Port == port
|
|
|
|
}
|
2014-04-13 16:29:52 +08:00
|
|
|
|
2012-09-26 16:55:56 +08:00
|
|
|
func (dn *DataNode) Url() string {
|
2012-12-04 13:27:57 +08:00
|
|
|
return dn.Ip + ":" + strconv.Itoa(dn.Port)
|
2012-09-26 16:55:56 +08:00
|
|
|
}
|
2012-09-17 08:31:15 +08:00
|
|
|
|
|
|
|
func (dn *DataNode) ToMap() interface{} {
|
|
|
|
ret := make(map[string]interface{})
|
2012-09-26 16:55:56 +08:00
|
|
|
ret["Url"] = dn.Url()
|
2012-09-17 08:31:15 +08:00
|
|
|
ret["PublicUrl"] = dn.PublicUrl
|
2021-02-19 17:38:56 +08:00
|
|
|
|
|
|
|
// aggregated volume info
|
|
|
|
var volumeCount, ecShardCount, maxVolumeCount int64
|
|
|
|
var volumeIds string
|
|
|
|
for _, diskUsage := range dn.diskUsages.usages {
|
|
|
|
volumeCount += diskUsage.volumeCount
|
|
|
|
ecShardCount += diskUsage.ecShardCount
|
|
|
|
maxVolumeCount += diskUsage.maxVolumeCount
|
|
|
|
}
|
|
|
|
|
|
|
|
for _, disk := range dn.Children() {
|
|
|
|
d := disk.(*Disk)
|
2021-02-20 06:22:36 +08:00
|
|
|
volumeIds += " " + d.GetVolumeIds()
|
2021-02-19 17:38:56 +08:00
|
|
|
}
|
|
|
|
|
|
|
|
ret["Volumes"] = volumeCount
|
|
|
|
ret["EcShards"] = ecShardCount
|
|
|
|
ret["Max"] = maxVolumeCount
|
2021-02-20 06:22:12 +08:00
|
|
|
ret["VolumeIds"] = volumeIds
|
2021-02-19 17:38:56 +08:00
|
|
|
|
2012-09-17 08:31:15 +08:00
|
|
|
return ret
|
2012-09-14 16:17:13 +08:00
|
|
|
}
|
2019-03-18 11:27:08 +08:00
|
|
|
|
|
|
|
func (dn *DataNode) ToDataNodeInfo() *master_pb.DataNodeInfo {
|
|
|
|
m := &master_pb.DataNodeInfo{
|
2021-02-16 19:03:00 +08:00
|
|
|
Id: string(dn.Id()),
|
|
|
|
DiskInfos: make(map[string]*master_pb.DiskInfo),
|
2019-03-18 11:27:08 +08:00
|
|
|
}
|
2021-02-16 18:47:02 +08:00
|
|
|
for _, c := range dn.Children() {
|
|
|
|
disk := c.(*Disk)
|
|
|
|
m.DiskInfos[string(disk.Id())] = disk.ToDiskInfo()
|
2019-05-23 13:44:28 +08:00
|
|
|
}
|
2019-03-18 11:27:08 +08:00
|
|
|
return m
|
|
|
|
}
|
2020-05-29 15:37:58 +08:00
|
|
|
|
|
|
|
// GetVolumeIds returns the human readable volume ids limited to count of max 100.
|
|
|
|
func (dn *DataNode) GetVolumeIds() string {
|
2020-07-08 19:57:19 +08:00
|
|
|
dn.RLock()
|
|
|
|
defer dn.RUnlock()
|
2021-02-16 18:47:02 +08:00
|
|
|
existingVolumes := dn.getVolumes()
|
|
|
|
ids := make([]int, 0, len(existingVolumes))
|
2020-05-29 15:37:58 +08:00
|
|
|
|
2021-02-16 18:47:02 +08:00
|
|
|
for k := range existingVolumes {
|
2020-05-29 15:37:58 +08:00
|
|
|
ids = append(ids, int(k))
|
|
|
|
}
|
|
|
|
|
2020-05-29 16:15:33 +08:00
|
|
|
return util.HumanReadableIntsMax(100, ids...)
|
2020-05-29 15:37:58 +08:00
|
|
|
}
|
2021-02-16 18:47:02 +08:00
|
|
|
|
|
|
|
func (dn *DataNode) getVolumes() []storage.VolumeInfo {
|
|
|
|
var existingVolumes []storage.VolumeInfo
|
|
|
|
for _, c := range dn.children {
|
|
|
|
disk := c.(*Disk)
|
|
|
|
existingVolumes = append(existingVolumes, disk.GetVolumes()...)
|
|
|
|
}
|
|
|
|
return existingVolumes
|
2021-02-16 19:03:00 +08:00
|
|
|
}
|