feature: creating vol with a specified zone

when creating vol which is not across zones,user specified a zone,to be created partition with the specified zone, if specified zone is not writable,we choose a zone randomly Signed-off-by: zhuhyc <[email protected]>
web-logs2 · Mar 13, 2020 · b91f267 · b91f267
1 parent 558e99f
commit b91f267
Show file tree

Hide file tree

Showing 14 changed files with 78 additions and 35 deletions.
diff --git a/cmd/master/scripts/manage_vol_create.sh b/cmd/master/scripts/manage_vol_create.sh
@@ -1,2 +1,2 @@
 #!/bin/bash
-curl -v "http://127.0.0.1/admin/createVol?name=test&capacity=100&owner=cfs&crossZone=false"
+curl -v "http://127.0.0.1/admin/createVol?name=test&capacity=100&owner=cfs&crossZone=false&zoneName=xx"
diff --git a/cmd/master/scripts/manage_cell_list.sh → cmd/master/scripts/manage_zone_list.sh b/cmd/master/scripts/manage_cell_list.sh → cmd/master/scripts/manage_zone_list.sh
diff --git a/cmd/master/scripts/manage_cell_update.sh → cmd/master/scripts/manage_zone_update.sh b/cmd/master/scripts/manage_cell_update.sh → cmd/master/scripts/manage_zone_update.sh
diff --git a/master/api_service.go b/master/api_service.go
@@ -546,9 +546,10 @@ func (m *Server) createVol(w http.ResponseWriter, r *http.Request) {
 		followerRead bool
 		authenticate bool
 		crossZone    bool
+		zoneName     string
 	)
 
-	if name, owner, mpCount, dpReplicaNum, size, capacity, followerRead, authenticate, crossZone, err = parseRequestToCreateVol(r); err != nil {
+	if name, owner, zoneName, mpCount, dpReplicaNum, size, capacity, followerRead, authenticate, crossZone, err = parseRequestToCreateVol(r); err != nil {
 		sendErrReply(w, r, &proto.HTTPReply{Code: proto.ErrCodeParamError, Msg: err.Error()})
 		return
 	}
@@ -557,7 +558,7 @@ func (m *Server) createVol(w http.ResponseWriter, r *http.Request) {
 		sendErrReply(w, r, &proto.HTTPReply{Code: proto.ErrCodeParamError, Msg: err.Error()})
 		return
 	}
-	if vol, err = m.cluster.createVol(name, owner, mpCount, dpReplicaNum, size, capacity, followerRead, authenticate, crossZone); err != nil {
+	if vol, err = m.cluster.createVol(name, owner, zoneName, mpCount, dpReplicaNum, size, capacity, followerRead, authenticate, crossZone); err != nil {
 		sendErrReply(w, r, newErrHTTPReply(err))
 		return
 	}
@@ -589,6 +590,7 @@ func newSimpleView(vol *Vol) *proto.SimpleVolView {
 		ID:                 vol.ID,
 		Name:               vol.Name,
 		Owner:              vol.Owner,
+		ZoneName:           vol.zoneName,
 		DpReplicaNum:       vol.dpReplicaNum,
 		MpReplicaNum:       vol.mpReplicaNum,
 		Status:             vol.Status,
@@ -1088,7 +1090,7 @@ func parseBoolFieldToUpdateVol(r *http.Request, vol *Vol) (followerRead, authent
 	return
 }
 
-func parseRequestToCreateVol(r *http.Request) (name, owner string, mpCount, dpReplicaNum, size, capacity int, followerRead, authenticate, crossZone bool, err error) {
+func parseRequestToCreateVol(r *http.Request) (name, owner, zoneName string, mpCount, dpReplicaNum, size, capacity int, followerRead, authenticate, crossZone bool, err error) {
 	if err = r.ParseForm(); err != nil {
 		return
 	}
@@ -1138,7 +1140,7 @@ func parseRequestToCreateVol(r *http.Request) (name, owner string, mpCount, dpRe
 	if crossZone, err = extractCrossZone(r); err != nil {
 		return
 	}
-
+	zoneName = r.FormValue(zoneNameKey)
 	return
 }
 

diff --git a/master/api_service_test.go b/master/api_service_test.go
@@ -95,7 +95,7 @@ func createDefaultMasterServerForTest() *Server {
 	testServer.cluster.checkMetaNodeHeartbeat()
 	time.Sleep(5 * time.Second)
 	testServer.cluster.scheduleToUpdateStatInfo()
-	vol, err := testServer.cluster.createVol(commonVolName, "cfs", 3, 3, 3, 100, false, false, false)
+	vol, err := testServer.cluster.createVol(commonVolName, "cfs", "",3, 3, 3, 100, false, false, false)
 	if err != nil {
 		panic(err)
 	}

diff --git a/master/cluster.go b/master/cluster.go
@@ -513,7 +513,7 @@ func (c *Cluster) createDataPartition(volName string, zoneNum int) (dp *DataPart
 	vol.createDpMutex.Lock()
 	defer vol.createDpMutex.Unlock()
 	errChannel := make(chan error, vol.dpReplicaNum)
-	if targetHosts, targetPeers, err = c.chooseTargetDataNodes("", nil, nil, int(vol.dpReplicaNum), zoneNum); err != nil {
+	if targetHosts, targetPeers, err = c.chooseTargetDataNodes("", nil, nil, int(vol.dpReplicaNum), zoneNum, vol.zoneName); err != nil {
 		goto errHandler
 	}
 	if partitionID, err = c.idAlloc.allocateDataPartitionID(); err != nil {
@@ -621,7 +621,7 @@ func (c *Cluster) decideZoneNum(crossZone bool) (zoneNum int) {
 	}
 	return zoneNum
 }
-func (c *Cluster) chooseTargetDataNodes(excludeZone string, excludeNodeSets []uint64, excludeHosts []string, replicaNum int, zoneNum int) (hosts []string, peers []proto.Peer, err error) {
+func (c *Cluster) chooseTargetDataNodes(excludeZone string, excludeNodeSets []uint64, excludeHosts []string, replicaNum int, zoneNum int, specifiedZone string) (hosts []string, peers []proto.Peer, err error) {
 
 	var (
 		masterZone *Zone
@@ -634,9 +634,22 @@ func (c *Cluster) chooseTargetDataNodes(excludeZone string, excludeNodeSets []ui
 	if replicaNum <= zoneNum {
 		zoneNum = replicaNum
 	}
-	zones, err = c.t.allocZonesForDataNode(zoneNum, replicaNum, excludeZones)
-	if err != nil {
-		return
+	// when creating vol,user specified a zone,we reset zoneNum to 1,to be created partition with specified zone,
+	//if specified zone is not writable,we choose a zone randomly
+	if specifiedZone != "" {
+		zoneNum = 1
+		zone, err := c.t.getZone(specifiedZone)
+		if err != nil {
+			Warn(c.Name, fmt.Sprintf("cluster[%v],specified zone[%v]is not writable", c.Name, specifiedZone))
+		} else {
+			zones = make([]*Zone, 0)
+			zones = append(zones, zone)
+		}
+	}
+	if zones == nil || specifiedZone == "" {
+		if zones, err = c.t.allocZonesForDataNode(zoneNum, replicaNum, excludeZones); err != nil {
+			return
+		}
 	}
 	//if vol enable cross zone,available zone less than 2,can't create partition
 	if zoneNum >= 2 && len(zones) < 2 {
@@ -849,7 +862,7 @@ func (c *Cluster) decommissionDataPartition(offlineAddr string, dp *DataPartitio
 			} else {
 				excludeZone = zones[0]
 			}
-			if targetHosts, _, err = c.chooseTargetDataNodes(excludeZone, excludeNodeSets, dp.Hosts, 1, 1); err != nil {
+			if targetHosts, _, err = c.chooseTargetDataNodes(excludeZone, excludeNodeSets, dp.Hosts, 1, 1, ""); err != nil {
 				goto errHandler
 			}
 		}
@@ -1253,7 +1266,7 @@ errHandler:
 
 // Create a new volume.
 // By default we create 3 meta partitions and 10 data partitions during initialization.
-func (c *Cluster) createVol(name, owner string, mpCount, dpReplicaNum, size, capacity int, followerRead, authenticate, crossZone bool) (vol *Vol, err error) {
+func (c *Cluster) createVol(name, owner, zoneName string, mpCount, dpReplicaNum, size, capacity int, followerRead, authenticate, crossZone bool) (vol *Vol, err error) {
 	var (
 		dataPartitionSize       uint64
 		readWriteDataPartitions int
@@ -1267,7 +1280,15 @@ func (c *Cluster) createVol(name, owner string, mpCount, dpReplicaNum, size, cap
 	if crossZone && c.t.zoneLen() <= 1 {
 		return nil, fmt.Errorf("cluster has one zone,can't cross zone")
 	}
-	if vol, err = c.doCreateVol(name, owner, dataPartitionSize, uint64(capacity), dpReplicaNum, followerRead, authenticate, crossZone); err != nil {
+	if crossZone && zoneName != "" {
+		return nil, fmt.Errorf("only the vol which don't across zones,can specified zoneName")
+	}
+	if zoneName != "" {
+		if _, err = c.t.getZone(zoneName); err != nil {
+			return
+		}
+	}
+	if vol, err = c.doCreateVol(name, owner, zoneName, dataPartitionSize, uint64(capacity), dpReplicaNum, followerRead, authenticate, crossZone); err != nil {
 		goto errHandler
 	}
 	if err = vol.initMetaPartitions(c, mpCount); err != nil {
@@ -1295,7 +1316,7 @@ errHandler:
 	return
 }
 
-func (c *Cluster) doCreateVol(name, owner string, dpSize, capacity uint64, dpReplicaNum int, followerRead, authenticate, crossZone bool) (vol *Vol, err error) {
+func (c *Cluster) doCreateVol(name, owner, zoneName string, dpSize, capacity uint64, dpReplicaNum int, followerRead, authenticate, crossZone bool) (vol *Vol, err error) {
 	var id uint64
 	c.createVolMutex.Lock()
 	defer c.createVolMutex.Unlock()
@@ -1307,7 +1328,7 @@ func (c *Cluster) doCreateVol(name, owner string, dpSize, capacity uint64, dpRep
 	if err != nil {
 		goto errHandler
 	}
-	vol = newVol(id, name, owner, dpSize, capacity, uint8(dpReplicaNum), defaultReplicaNum, followerRead, authenticate, crossZone)
+	vol = newVol(id, name, owner, zoneName, dpSize, capacity, uint8(dpReplicaNum), defaultReplicaNum, followerRead, authenticate, crossZone)
 	// refresh oss secure
 	vol.refreshOSSSecure()
 	if err = c.syncAddVol(vol); err != nil {
@@ -1356,7 +1377,7 @@ func (c *Cluster) updateInodeIDRange(volName string, start uint64) (err error) {
 }
 
 // Choose the target hosts from the available zones and meta nodes.
-func (c *Cluster) chooseTargetMetaHosts(excludeZone string, excludeNodeSets []uint64, excludeHosts []string, replicaNum int, crossZone bool) (hosts []string, peers []proto.Peer, err error) {
+func (c *Cluster) chooseTargetMetaHosts(excludeZone string, excludeNodeSets []uint64, excludeHosts []string, replicaNum int, crossZone bool, specifiedZone string) (hosts []string, peers []proto.Peer, err error) {
 	var (
 		zones      []*Zone
 		masterZone *Zone
@@ -1369,10 +1390,24 @@ func (c *Cluster) chooseTargetMetaHosts(excludeZone string, excludeNodeSets []ui
 	if replicaNum < zoneNum {
 		zoneNum = replicaNum
 	}
-	zones, err = c.t.allocZonesForMetaNode(zoneNum, replicaNum, excludeZones)
-	if err != nil {
-		return
+	// when creating vol,user specified a zone,we reset zoneNum to 1,to be created partition with specified zone,
+	//if specified zone is not writable,we choose a zone randomly
+	if specifiedZone != "" {
+		zoneNum = 1
+		zone, err := c.t.getZone(specifiedZone)
+		if err != nil {
+			Warn(c.Name, fmt.Sprintf("cluster[%v],specified zone[%v]is not writable", c.Name, specifiedZone))
+		} else {
+			zones = make([]*Zone, 0)
+			zones = append(zones, zone)
+		}
 	}
+	if zones == nil || specifiedZone == "" {
+		if zones, err = c.t.allocZonesForMetaNode(zoneNum, replicaNum, excludeZones); err != nil {
+			return
+		}
+	}
+
 	if crossZone && len(zones) < 2 {
 		log.LogWarn(fmt.Sprintf("action[chooseTargetMetaNodes] ,no enough zones [%v] to be selected, expect select [%v] zones", len(zones), zoneNum))
 		return nil, nil, fmt.Errorf("action[chooseTargetMetaNodes] no enough zones [%v] to be selected, expect select [%v] zones", len(zones), zoneNum)

diff --git a/master/cluster_task.go b/master/cluster_task.go
@@ -134,7 +134,7 @@ func (c *Cluster) decommissionMetaPartition(nodeAddr string, mp *MetaPartition)
 				excludeZone = zones[0]
 			}
 			// choose a meta node in other zone
-			if _, newPeers, err = c.chooseTargetMetaHosts(excludeZone, excludeNodeSets, oldHosts, 1, false); err != nil {
+			if _, newPeers, err = c.chooseTargetMetaHosts(excludeZone, excludeNodeSets, oldHosts, 1, false, ""); err != nil {
 				goto errHandler
 			}
 		}

diff --git a/master/cluster_test.go b/master/cluster_test.go
@@ -19,7 +19,7 @@ func buildPanicVol() *Vol {
 	if err != nil {
 		return nil
 	}
-	vol := newVol(id, commonVol.Name, commonVol.Owner, commonVol.dataPartitionSize, commonVol.Capacity, defaultReplicaNum, defaultReplicaNum, false, false,false)
+	vol := newVol(id, commonVol.Name, commonVol.Owner,"", commonVol.dataPartitionSize, commonVol.Capacity, defaultReplicaNum, defaultReplicaNum, false, false,false)
 	vol.dataPartitions = nil
 	return vol
 }

diff --git a/master/metadata_fsm_op.go b/master/metadata_fsm_op.go
@@ -117,6 +117,7 @@ type volValue struct {
 	FollowerRead      bool
 	Authenticate      bool
 	CrossZone         bool
+	ZoneName          string
 	OSSAccessKey      string
 	OSSSecretKey      string
 }
@@ -139,6 +140,7 @@ func newVolValue(vol *Vol) (vv *volValue) {
 		FollowerRead:      vol.FollowerRead,
 		Authenticate:      vol.authenticate,
 		CrossZone:         vol.crossZone,
+		ZoneName:          vol.zoneName,
 		OSSAccessKey:      vol.OSSAccessKey,
 		OSSSecretKey:      vol.OSSSecretKey,
 	}
@@ -186,16 +188,16 @@ func newMetaNodeValue(metaNode *MetaNode) *metaNodeValue {
 }
 
 type nodeSetValue struct {
-	ID          uint64
-	Capacity    int
-	ZoneName    string
+	ID       uint64
+	Capacity int
+	ZoneName string
 }
 
 func newNodeSetValue(nset *nodeSet) (nsv *nodeSetValue) {
 	nsv = &nodeSetValue{
-		ID:          nset.ID,
-		Capacity:    nset.Capacity,
-		ZoneName:    nset.zoneName,
+		ID:       nset.ID,
+		Capacity: nset.Capacity,
+		ZoneName: nset.zoneName,
 	}
 	return
 }

diff --git a/master/topology.go b/master/topology.go
@@ -679,7 +679,7 @@ func (zone *Zone) getAvailMetaNodeHosts(excludeNodeSets []uint64, excludeHosts [
 	}
 	ns, err := zone.allocNodeSetForMetaNode(excludeNodeSets, uint8(replicaNum))
 	if err != nil {
-		return
+		return nil, nil, errors.NewErrorf("zone[%v],err[%v]", zone.name, err)
 	}
 	return ns.getAvailMetaNodeHosts(excludeHosts, replicaNum)
 

diff --git a/master/topology_test.go b/master/topology_test.go
@@ -111,13 +111,13 @@ func TestAllocZones(t *testing.T) {
 	cluster.t = topo
 	cluster.cfg = newClusterConfig()
 	//don't cross zone
-	hosts, _, err := cluster.chooseTargetDataNodes("", nil, nil, replicaNum,1)
+	hosts, _, err := cluster.chooseTargetDataNodes("", nil, nil, replicaNum, 1, "")
 	if err != nil {
 		t.Error(err)
 		return
 	}
 	//cross zone
-	hosts, _, err = cluster.chooseTargetDataNodes("", nil, nil, replicaNum,2)
+	hosts, _, err = cluster.chooseTargetDataNodes("", nil, nil, replicaNum, 2, "")
 	if err != nil {
 		t.Error(err)
 		return

diff --git a/master/vol.go b/master/vol.go
@@ -41,6 +41,7 @@ type Vol struct {
 	FollowerRead       bool
 	authenticate       bool
 	crossZone          bool
+	zoneName           string
 	MetaPartitions     map[uint64]*MetaPartition
 	mpsLock            sync.RWMutex
 	dataPartitions     *DataPartitionMap
@@ -51,7 +52,7 @@ type Vol struct {
 	sync.RWMutex
 }
 
-func newVol(id uint64, name, owner string, dpSize, capacity uint64, dpReplicaNum, mpReplicaNum uint8, followerRead, authenticate, crossZone bool) (vol *Vol) {
+func newVol(id uint64, name, owner, zoneName string, dpSize, capacity uint64, dpReplicaNum, mpReplicaNum uint8, followerRead, authenticate, crossZone bool) (vol *Vol) {
 	vol = &Vol{ID: id, Name: name, MetaPartitions: make(map[uint64]*MetaPartition, 0)}
 	vol.dataPartitions = newDataPartitionMap(name)
 	if dpReplicaNum <= 1 {
@@ -75,6 +76,7 @@ func newVol(id uint64, name, owner string, dpSize, capacity uint64, dpReplicaNum
 	vol.FollowerRead = followerRead
 	vol.authenticate = authenticate
 	vol.crossZone = crossZone
+	vol.zoneName = zoneName
 	vol.viewCache = make([]byte, 0)
 	vol.mpsCache = make([]byte, 0)
 	return
@@ -85,6 +87,7 @@ func newVolFromVolValue(vv *volValue) (vol *Vol) {
 		vv.ID,
 		vv.Name,
 		vv.Owner,
+		vv.ZoneName,
 		vv.DataPartitionSize,
 		vv.Capacity,
 		vv.DpReplicaNum,
@@ -691,7 +694,7 @@ func (vol *Vol) doCreateMetaPartition(c *Cluster, start, end uint64) (mp *MetaPa
 		wg          sync.WaitGroup
 	)
 	errChannel := make(chan error, vol.mpReplicaNum)
-	if hosts, peers, err = c.chooseTargetMetaHosts("", nil, nil, int(vol.mpReplicaNum), vol.crossZone); err != nil {
+	if hosts, peers, err = c.chooseTargetMetaHosts("", nil, nil, int(vol.mpReplicaNum), vol.crossZone, vol.zoneName); err != nil {
 		log.LogErrorf("action[doCreateMetaPartition] chooseTargetMetaHosts err[%v]", err)
 		return nil, errors.NewError(err)
 	}

diff --git a/master/vol_test.go b/master/vol_test.go
@@ -171,7 +171,7 @@ func markDeleteVol(name string, t *testing.T) {
 
 func TestVolReduceReplicaNum(t *testing.T) {
 	volName := "reduce-replica-num"
-	vol, err := server.cluster.createVol(volName, volName, 3, 3, util.DefaultDataPartitionSize, 100, false, false,false)
+	vol, err := server.cluster.createVol(volName, volName, "",3, 3, util.DefaultDataPartitionSize, 100, false, false,false)
 	if err != nil {
 		t.Error(err)
 		return
@@ -210,7 +210,7 @@ func TestVolReduceReplicaNum(t *testing.T) {
 func TestConcurrentReadWriteDataPartitionMap(t *testing.T) {
 	name := "TestConcurrentReadWriteDataPartitionMap"
 	var volID uint64 = 1
-	vol := newVol(volID, name, name, util.DefaultDataPartitionSize, 100, defaultReplicaNum, defaultReplicaNum, false, false,false)
+	vol := newVol(volID, name, name,"", util.DefaultDataPartitionSize, 100, defaultReplicaNum, defaultReplicaNum, false, false,false)
 	//unavaliable mp
 	mp1 := newMetaPartition(1, 1, defaultMaxMetaPartitionInodeID, 3, name, volID)
 	vol.addMetaPartition(mp1)

diff --git a/proto/admin_proto.go b/proto/admin_proto.go
@@ -403,6 +403,7 @@ type SimpleVolView struct {
 	ID                 uint64
 	Name               string
 	Owner              string
+	ZoneName           string
 	DpReplicaNum       uint8
 	MpReplicaNum       uint8
 	Status             uint8