Merge branch 'master' into acl

client/fs/file.go

@@ -80,6 +80,9 @@ func (f *File) Attr(ctx context.Context, a *fuse.Attr) error {
 	if gen >= info.Generation {
 		a.Size = uint64(fileSize)
 	}
+	if proto.IsSymlink(info.Mode) {
+		a.Size = uint64(len(info.Target))
+	}
 
 	log.LogDebugf("TRACE Attr: inode(%v) attr(%v)", info, a)
 	return nil

sdk/data/stream/stream_conn.go

@@ -57,11 +57,12 @@ func NewStreamConn(dp *wrapper.DataPartition, follower bool) *StreamConn {
 	}
 
 	epoch := atomic.AddUint64(&dp.Epoch, 1)
-	choice := len(dp.Hosts)
+	hosts := sortByStatus(dp, false)
+	choice := len(hosts)
 	currAddr := dp.LeaderAddr
 	if choice > 0 {
 		index := int(epoch) % choice
-		currAddr = dp.Hosts[index]
+		currAddr = hosts[index]
 	}
 
 	return &StreamConn{
@@ -106,7 +107,9 @@ func (sc *StreamConn) sendToPartition(req *Packet, getReply GetReplyFunc) (err e
 		log.LogWarnf("sendToPartition: get connection to curr addr failed, addr(%v) reqPacket(%v) err(%v)", sc.currAddr, req, err)
 	}
 
-	for _, addr := range sc.dp.Hosts {
+	hosts := sortByStatus(sc.dp, true)
+
+	for _, addr := range hosts {
 		log.LogWarnf("sendToPartition: try addr(%v) reqPacket(%v)", addr, req)
 		conn, err = StreamConnPool.GetConnect(addr)
 		if err != nil {
@@ -155,3 +158,30 @@ func (sc *StreamConn) sendToConn(conn *net.TCPConn, req *Packet, getReply GetRep
 	log.LogDebugf("sendToConn exit: send to addr(%v) reqPacket(%v) err(%v)", sc.currAddr, req, err)
 	return
 }
+
+// sortByStatus will return hosts list sort by host status for DataPartition.
+// If param selectAll is true, hosts with status(true) is in front and hosts with status(false) is in behind.
+// If param selectAll is false, only return hosts with status(true).
+func sortByStatus(dp *wrapper.DataPartition, selectAll bool) (hosts []string) {
+	var failedHosts []string
+	hostsStatus := dp.ClientWrapper.HostsStatus
+	for _, addr := range dp.Hosts {
+		status, ok := hostsStatus[addr]
+		if ok {
+			if status {
+				hosts = append(hosts, addr)
+			} else {
+				failedHosts = append(failedHosts, addr)
+			}
+		} else {
+			failedHosts = append(failedHosts, addr)
+			log.LogWarnf("sortByStatus: can not find host[%v] in HostsStatus, dp[%d]", addr, dp.PartitionID)
+		}
+	}
+
+	if selectAll {
+		hosts = append(hosts, failedHosts...)
+	}
+
+	return
+}

sdk/data/wrapper/data_partition.go

@@ -31,6 +31,7 @@ type DataPartition struct {
 	proto.DataPartitionResponse
 	RandomWrite   bool
 	PartitionType string
+	ClientWrapper *Wrapper
 	Metrics       *DataPartitionMetrics
 }
 

sdk/data/wrapper/wrapper.go

@@ -49,6 +49,8 @@ type Wrapper struct {
 	mc                    *masterSDK.MasterClient
 	stopOnce              sync.Once
 	stopC                 chan struct{}
+
+	HostsStatus map[string]bool
 }
 
 // NewDataPartitionWrapper returns a new data partition wrapper.
@@ -60,6 +62,7 @@ func NewDataPartitionWrapper(volName string, masters []string) (w *Wrapper, err
 	w.volName = volName
 	w.rwPartition = make([]*DataPartition, 0)
 	w.partitions = make(map[uint64]*DataPartition)
+	w.HostsStatus = make(map[string]bool)
 	if err = w.updateClusterInfo(); err != nil {
 		err = errors.Trace(err, "NewDataPartitionWrapper:")
 		return
@@ -72,6 +75,9 @@ func NewDataPartitionWrapper(volName string, masters []string) (w *Wrapper, err
 		err = errors.Trace(err, "NewDataPartitionWrapper:")
 		return
 	}
+	if err = w.updateDataNodeStatus(); err != nil {
+		log.LogErrorf("NewDataPartitionWrapper: init DataNodeStatus failed, [%v]", err)
+	}
 	go w.update()
 	return
 }
@@ -120,6 +126,7 @@ func (w *Wrapper) update() {
 		select {
 		case <-ticker.C:
 			w.updateDataPartition(false)
+			w.updateDataNodeStatus()
 		case <-w.stopC:
 			return
 		}
@@ -136,7 +143,10 @@ func (w *Wrapper) updateDataPartition(isInit bool) (err error) {
 	log.LogInfof("updateDataPartition: get data partitions: volume(%v) partitions(%v)", w.volName, len(dpv.DataPartitions))
 
 	var convert = func(response *proto.DataPartitionResponse) *DataPartition {
-		return &DataPartition{DataPartitionResponse: *response}
+		return &DataPartition{
+			DataPartitionResponse: *response,
+			ClientWrapper:         w,
+		}
 	}
 
 	rwPartitionGroups := make([]*DataPartition, 0)
@@ -251,3 +261,22 @@ func (w *Wrapper) GetDataPartition(partitionID uint64) (*DataPartition, error) {
 func (w *Wrapper) WarningMsg() string {
 	return fmt.Sprintf("%s_client_warning", w.clusterName)
 }
+
+func (w *Wrapper) updateDataNodeStatus() (err error) {
+	var cv *proto.ClusterView
+	cv, err = w.mc.AdminAPI().GetCluster()
+	if err != nil {
+		log.LogErrorf("updateDataNodeStatus: get cluster fail: err(%v)", err)
+		return
+	}
+
+	newHostsStatus := make(map[string]bool)
+	for _, node := range cv.DataNodes {
+		newHostsStatus[node.Addr] = node.Status
+	}
+	log.LogInfof("updateDataNodeStatus: update %d hosts status", len(newHostsStatus))
+
+	w.HostsStatus = newHostsStatus
+
+	return
+}