volume growth: update volume growth test, and fix bugs

go/topology/node.go

@@ -2,7 +2,6 @@ package topology
 
 import (
 	"errors"
-	"fmt"
 	"math/rand"
 
 	"sort"
@@ -151,7 +150,7 @@ func (n *NodeImpl) GetValue() interface{} {
 func (n *NodeImpl) ReserveOneVolume(r int) (assignedNode *DataNode, err error) {
 	for _, node := range n.children {
 		freeSpace := node.FreeSpace()
-		fmt.Println("r =", r, ", node =", node, ", freeSpace =", freeSpace)
+		// fmt.Println("r =", r, ", node =", node, ", freeSpace =", freeSpace)
 		if freeSpace <= 0 {
 			continue
 		}
@@ -159,7 +158,7 @@ func (n *NodeImpl) ReserveOneVolume(r int) (assignedNode *DataNode, err error) {
 			r -= freeSpace
 		} else {
 			if node.IsDataNode() && node.FreeSpace() > 0 {
-				fmt.Println("assigned to node =", node, ", freeSpace =", node.FreeSpace())
+				// fmt.Println("assigned to node =", node, ", freeSpace =", node.FreeSpace())
 				return node.(*DataNode), nil
 			}
 			assignedNode, err = node.ReserveOneVolume(r)

go/topology/volume_growth.go

@@ -159,7 +159,7 @@ func makeExceptNodeFilter(nodes []Node) FilterNodeFn {
 // 2.2 collect all racks that have rp.SameRackCount+1
 // 2.2 collect all data centers that have DiffRackCount+rp.SameRackCount+1
 // 2. find rest data nodes
-func (vg *VolumeGrowth) findEmptySlotsForOneVolume(topo *Topology, option *VolumeGrowOption, existsServer *VolumeLocationList) (additionServers []*DataNode, err error) {
+func (vg *VolumeGrowth) findEmptySlotsForOneVolume(topo *Topology, option *VolumeGrowOption, existsServers *VolumeLocationList) (additionServers []*DataNode, err error) {
 	//find main datacenter and other data centers
 	pickNodesFn := PickLowUsageNodeFn
 	rp := option.ReplicaPlacement
@@ -194,8 +194,8 @@ func (vg *VolumeGrowth) findEmptySlotsForOneVolume(topo *Topology, option *Volum
 		return mainNode, restNodes, nil
 	}
 	var existsNode []Node
-	if existsServer != nil {
-		existsNode = existsServer.DiffDataCenters()
+	if existsServers != nil {
+		existsNode = existsServers.DiffDataCenters()
 	}
 	mainDataCenter, otherDataCenters, dc_err := pickMainAndRestNodes(topo, rp.DiffDataCenterCount+1,
 		func(node Node) error {
@@ -205,8 +205,8 @@ func (vg *VolumeGrowth) findEmptySlotsForOneVolume(topo *Topology, option *Volum
 		return nil, dc_err
 	}
 	//find main rack and other racks
-	if existsServer != nil {
-		existsNode = existsServer.DiffRacks(mainDataCenter.(*DataCenter))
+	if existsServers != nil {
+		existsNode = existsServers.DiffRacks(mainDataCenter.(*DataCenter))
 	} else {
 		existsNode = nil
 	}
@@ -221,8 +221,8 @@ func (vg *VolumeGrowth) findEmptySlotsForOneVolume(topo *Topology, option *Volum
 	}
 
 	//find main server and other servers
-	if existsServer != nil {
-		existsNode = existsServer.SameServers(mainRack.(*Rack))
+	if existsServers != nil {
+		existsNode = existsServers.SameServers(mainRack.(*Rack))
 	} else {
 		existsNode = nil
 	}
@@ -242,8 +242,11 @@ func (vg *VolumeGrowth) findEmptySlotsForOneVolume(topo *Topology, option *Volum
 	if server_err != nil {
 		return nil, server_err
 	}
+	if existsServers != nil && existsServers.ContainsDataNode(mainServer.(*DataNode)) {
+	} else {
+		additionServers = append(additionServers, mainServer.(*DataNode))
+	}
 
-	additionServers = append(additionServers, mainServer.(*DataNode))
 	for _, server := range otherServers {
 		additionServers = append(additionServers, server.(*DataNode))
 	}

go/topology/volume_growth_test.go

@@ -5,8 +5,11 @@ import (
 	"fmt"
 	"testing"
 
+	"strings"
+
 	"github.com/chrislusf/seaweedfs/go/sequence"
 	"github.com/chrislusf/seaweedfs/go/storage"
+	"github.com/syndtr/goleveldb/leveldb/errors"
 )
 
 var topologyLayout = `
@@ -100,6 +103,14 @@ var topologyLayout = `
 }
 `
 
+var testLocList = [][]string{
+	{"server111", "server121"},
+	{"server111", "server112"},
+	{"server111", "server112", "server113"},
+	{"server111", "server221", "server321"},
+	{"server112"},
+}
+
 func setup(topologyLayout string) *Topology {
 	var data interface{}
 	err := json.Unmarshal([]byte(topologyLayout), &data)
@@ -163,4 +174,65 @@ func TestFindEmptySlotsForOneVolume(t *testing.T) {
 	for _, server := range servers {
 		fmt.Printf("assigned node: %s, free space: %d\n", server.Id(), server.FreeSpace())
 	}
+
+}
+
+func getDataNodeFromId(topo *Topology, id string) (foundDn *DataNode) {
+	nid := NodeId(id)
+	topo.WalkDataNode(func(dn *DataNode) (e error) {
+		if dn.Id() == nid {
+			foundDn = dn
+			e = errors.New("Found.")
+		}
+		return
+	})
+	return
+}
+
+func setupTestLocationList(topo *Topology) (ret []*VolumeLocationList) {
+
+	for _, ll := range testLocList {
+		vl := &VolumeLocationList{}
+		for _, nid := range ll {
+			if n := getDataNodeFromId(topo, nid); n != nil {
+				vl.list = append(vl.list, n)
+			}
+		}
+		ret = append(ret, vl)
+	}
+	return
+}
+
+func joinNodeId(dns []*DataNode) string {
+	ss := []string{}
+	for _, dn := range dns {
+		ss = append(ss, string(dn.Id()))
+	}
+	return strings.Join(ss, ", ")
+}
+
+func TestFindEmptySlotsWithExistsNodes(t *testing.T) {
+	topo := setup(topologyLayout)
+	vg := NewDefaultVolumeGrowth()
+	rp, _ := storage.NewReplicaPlacementFromString("112")
+	volumeGrowOption := &VolumeGrowOption{
+		Collection:       "",
+		ReplicaPlacement: rp,
+		DataCenter:       "dc1",
+		Rack:             "",
+		DataNode:         "",
+	}
+	testLocationList := setupTestLocationList(topo)
+	for _, locationList := range testLocationList {
+		lrp := locationList.CalcReplicaPlacement()
+		t.Logf("location list: [%s], replica placement = %s\n", joinNodeId(locationList.list), lrp.String())
+		if lrp.Compare(rp) < 0 {
+			servers, err := vg.findEmptySlotsForOneVolume(topo, volumeGrowOption, locationList)
+			if err != nil {
+				t.Log("finding empty slots error :", err)
+				t.Fail()
+			}
+			t.Logf("assigned node: %s\n\n", joinNodeId(servers))
+		}
+	}
 }

go/topology/volume_location_list.go

@@ -118,7 +118,9 @@ func (dnll *VolumeLocationList) DiffRacks(mainDC *DataCenter) []Node {
 		racks = append(racks, mainRack)
 	}
 	for rack := range m {
-		racks = append(racks, rack)
+		if rack != mainRack {
+			racks = append(racks, rack)
+		}
 	}
 	return racks
 }
@@ -138,17 +140,28 @@ func (dnll *VolumeLocationList) SameServers(mainRack *Rack) (servers []Node) {
 }
 
 func (dnll *VolumeLocationList) CalcReplicaPlacement() (rp *storage.ReplicaPlacement) {
-	dcs := dnll.DiffDataCenters()
-	rs := dnll.DiffRacks(dcs[0].(*DataCenter))
-	ss := dnll.SameServers(rs[0].(*Rack))
+	var dcs, rs, ss []Node
+	dcs = dnll.DiffDataCenters()
+	if len(dcs) > 0 {
+		rs = dnll.DiffRacks(dcs[0].(*DataCenter))
+		if len(rs) > 0 {
+			ss = dnll.SameServers(rs[0].(*Rack))
+		}
+	}
+
 	rp = &storage.ReplicaPlacement{
-		len(dcs) - 1,
-		len(rs) - 1,
-		len(ss) - 1,
+		SameRackCount:       len(ss) - 1,
+		DiffRackCount:       len(rs) - 1,
+		DiffDataCenterCount: len(dcs) - 1,
 	}
 	return
 }
 
-//func (dnll *VolumeLocationList)ContainDataNode(nodeType, id string)bool {
-//
-//}
+func (dnll *VolumeLocationList) ContainsDataNode(n *DataNode) bool {
+	for _, dn := range dnll.list {
+		if dn == n {
+			return true
+		}
+	}
+	return false
+}