Skip to content

Commit

Permalink
fix: resolve cycle import issue in unit test by move common entity st…
Browse files Browse the repository at this point in the history 
…ruct to proto package.

fix: make all unit test passed

Signed-off-by: Mofei Zhang <[email protected]>
  • Loading branch information
@mervinkid
mervinkid committed Dec 25, 2019
1 parent bfa731c commit 90daee7
Show file tree
Hide file tree
Showing 20 changed files with 403 additions and 163 deletions.
12 changes: 6 additions & 6 deletions datanode/partition.go
View file
Original file line number Diff line number Diff line change
Expand Up @@ -26,7 +26,11 @@ import (
"sync"
"time"

"github.com/chubaofs/chubaofs/master"
"hash/crc32"
"net"
"sort"
"syscall"

"github.com/chubaofs/chubaofs/proto"
"github.com/chubaofs/chubaofs/raftstore"
"github.com/chubaofs/chubaofs/repl"
Expand All @@ -35,10 +39,6 @@ import (
"github.com/chubaofs/chubaofs/util/exporter"
"github.com/chubaofs/chubaofs/util/log"
raftProto "github.com/tiglabs/raft/proto"
"hash/crc32"
"net"
"sort"
"syscall"
)

const (
Expand Down Expand Up @@ -587,7 +587,7 @@ func (dp *DataPartition) compareReplicas(v1, v2 []string) (equals bool) {
// Fetch the replica information from the master.
func (dp *DataPartition) fetchReplicasFromMaster() (isLeader bool, replicas []string, err error) {

var partition *master.DataPartition
var partition *proto.DataPartitionInfo
if partition, err = MasterClient.AdminAPI().GetDataPartition(dp.volumeID, dp.partitionID); err != nil {
isLeader = false
return
Expand Down
15 changes: 7 additions & 8 deletions datanode/server.go
View file
Original file line number Diff line number Diff line change
Expand Up @@ -16,7 +16,6 @@ package datanode

import (
"fmt"
"github.com/chubaofs/chubaofs/master"
"net"
"net/http"
"regexp"
Expand Down Expand Up @@ -89,10 +88,10 @@ type DataNode struct {
raftReplica string
raftStore raftstore.RaftStore

tcpListener net.Listener
stopC chan bool
state uint32
wg sync.WaitGroup
tcpListener net.Listener
stopC chan bool
state uint32
wg sync.WaitGroup
}

func NewServer() *DataNode {
Expand Down Expand Up @@ -259,7 +258,7 @@ func (s *DataNode) startSpaceManager(cfg *config.Config) (err error) {
// The startup of a data node will be blocked until the registration succeeds.
func (s *DataNode) register(cfg *config.Config) {
var (
err error
err error
)

timer := time.NewTimer(0)
Expand Down Expand Up @@ -313,13 +312,13 @@ type DataNodeInfo struct {
}

func (s *DataNode) checkLocalPartitionMatchWithMaster() (err error) {
var convert = func(node *master.DataNode) *DataNodeInfo {
var convert = func(node *proto.DataNodeInfo) *DataNodeInfo {
result := &DataNodeInfo{}
result.Addr = node.Addr
result.PersistenceDataPartitions = node.PersistenceDataPartitions
return result
}
var dataNode *master.DataNode
var dataNode *proto.DataNodeInfo
for i := 0; i < 3; i++ {
if dataNode, err = MasterClient.NodeAPI().GetDataNode(s.localServerAddr); err != nil {
log.LogErrorf("checkLocalPartitionMatchWithMaster error %v", err)
Expand Down
101 changes: 82 additions & 19 deletions master/api_service.go
View file
Original file line number Diff line number Diff line change
Expand Up @@ -90,10 +90,7 @@ func newCellView() *CellView {
return &CellView{DataNodes: make([]NodeView, 0)}
}

type badPartitionView struct {
DiskPath string
PartitionIDs []uint64
}
type badPartitionView = proto.BadPartitionView

// Set the threshold of the memory usage on each meta node.
// If the memory usage reaches this threshold, then all the mata partition will be marked as readOnly.
Expand Down Expand Up @@ -165,7 +162,7 @@ func (m *Server) getTopology(w http.ResponseWriter, r *http.Request) {
}

func (m *Server) getCluster(w http.ResponseWriter, r *http.Request) {
cv := &ClusterView{
cv := &proto.ClusterView{
Name: m.cluster.Name,
LeaderAddr: m.leaderInfo.addr,
DisableAutoAlloc: m.cluster.DisableAutoAllocate,
Expand All @@ -174,10 +171,10 @@ func (m *Server) getCluster(w http.ResponseWriter, r *http.Request) {
MaxDataPartitionID: m.cluster.idAlloc.dataPartitionID,
MaxMetaNodeID: m.cluster.idAlloc.commonID,
MaxMetaPartitionID: m.cluster.idAlloc.metaPartitionID,
MetaNodes: make([]NodeView, 0),
DataNodes: make([]NodeView, 0),
VolStatInfo: make([]*volStatInfo, 0),
BadPartitionIDs: make([]badPartitionView, 0),
MetaNodes: make([]proto.NodeView, 0),
DataNodes: make([]proto.NodeView, 0),
VolStatInfo: make([]*proto.VolStatInfo, 0),
BadPartitionIDs: make([]proto.BadPartitionView, 0),
}

vols := m.cluster.allVolNames()
Expand Down Expand Up @@ -293,7 +290,7 @@ func (m *Server) getDataPartition(w http.ResponseWriter, r *http.Request) {
}
}

sendOkReply(w, r, newSuccessHTTPReply(dp))
sendOkReply(w, r, newSuccessHTTPReply(dp.ToProto()))
}

// Load the data partition.
Expand Down Expand Up @@ -597,9 +594,10 @@ func (m *Server) addDataNode(w http.ResponseWriter, r *http.Request) {

func (m *Server) getDataNode(w http.ResponseWriter, r *http.Request) {
var (
nodeAddr string
dataNode *DataNode
err error
nodeAddr string
dataNode *DataNode
dataNodeInfo *proto.DataNodeInfo
err error
)
if nodeAddr, err = parseAndExtractNodeAddr(r); err != nil {
sendErrReply(w, r, &proto.HTTPReply{Code: proto.ErrCodeParamError, Msg: err.Error()})
Expand All @@ -611,7 +609,27 @@ func (m *Server) getDataNode(w http.ResponseWriter, r *http.Request) {
return
}
dataNode.PersistenceDataPartitions = m.cluster.getAllDataPartitionIDByDatanode(nodeAddr)
sendOkReply(w, r, newSuccessHTTPReply(dataNode))

dataNodeInfo = &proto.DataNodeInfo{
Total: dataNode.Total,
Used: dataNode.Used,
AvailableSpace: dataNode.AvailableSpace,
ID: dataNode.ID,
CellName: dataNode.CellName,
Addr: dataNode.Addr,
ReportTime: dataNode.ReportTime,
IsActive: dataNode.isActive,
UsageRatio: dataNode.UsageRatio,
SelectedTimes: dataNode.SelectedTimes,
Carry: dataNode.Carry,
DataPartitionReports: dataNode.DataPartitionReports,
DataPartitionCount: dataNode.DataPartitionCount,
NodeSetID: dataNode.NodeSetID,
PersistenceDataPartitions: dataNode.PersistenceDataPartitions,
BadDisks: dataNode.BadDisks,
}

sendOkReply(w, r, newSuccessHTTPReply(dataNodeInfo))
}

// Decommission a data node. This will decommission all the data partition on that node.
Expand Down Expand Up @@ -710,9 +728,10 @@ func (m *Server) addMetaNode(w http.ResponseWriter, r *http.Request) {

func (m *Server) getMetaNode(w http.ResponseWriter, r *http.Request) {
var (
nodeAddr string
metaNode *MetaNode
err error
nodeAddr string
metaNode *MetaNode
metaNodeInfo *proto.MetaNodeInfo
err error
)
if nodeAddr, err = parseAndExtractNodeAddr(r); err != nil {
sendErrReply(w, r, &proto.HTTPReply{Code: proto.ErrCodeParamError, Msg: err.Error()})
Expand All @@ -724,7 +743,24 @@ func (m *Server) getMetaNode(w http.ResponseWriter, r *http.Request) {
return
}
metaNode.PersistenceMetaPartitions = m.cluster.getAllMetaPartitionIDByMetaNode(nodeAddr)
sendOkReply(w, r, newSuccessHTTPReply(metaNode))
metaNodeInfo = &proto.MetaNodeInfo{
ID: metaNode.ID,
Addr: metaNode.Addr,
IsActive: metaNode.IsActive,
CellName: metaNode.CellName,
MaxMemAvailWeight: metaNode.MaxMemAvailWeight,
Total: metaNode.Total,
Used: metaNode.Used,
Ratio: metaNode.Ratio,
SelectCount: metaNode.SelectCount,
Carry: metaNode.Carry,
Threshold: metaNode.Threshold,
ReportTime: metaNode.ReportTime,
MetaPartitionCount: metaNode.MetaPartitionCount,
NodeSetID: metaNode.NodeSetID,
PersistenceMetaPartitions: metaNode.PersistenceMetaPartitions,
}
sendOkReply(w, r, newSuccessHTTPReply(metaNodeInfo))
}

func (m *Server) decommissionMetaPartition(w http.ResponseWriter, r *http.Request) {
Expand Down Expand Up @@ -1472,7 +1508,34 @@ func (m *Server) getMetaPartition(w http.ResponseWriter, r *http.Request) {
sendErrReply(w, r, newErrHTTPReply(proto.ErrMetaPartitionNotExists))
return
}
sendOkReply(w, r, newSuccessHTTPReply(mp))

var toInfo = func(mp *MetaPartition) *proto.MetaPartitionInfo {
var replicas = make([]*proto.MetaReplicaInfo, len(mp.Replicas))
for i := 0; i < len(replicas); i++ {
replicas[i] = &proto.MetaReplicaInfo{
Addr: mp.Replicas[i].Addr,
ReportTime: mp.Replicas[i].ReportTime,
Status: mp.Replicas[i].Status,
IsLeader: mp.Replicas[i].IsLeader,
}
}
var mpInfo = &proto.MetaPartitionInfo{
PartitionID: mp.PartitionID,
Start: mp.Start,
End: mp.End,
MaxInodeID: mp.MaxInodeID,
Replicas: replicas,
ReplicaNum: mp.ReplicaNum,
Status: mp.Status,
Hosts: mp.Hosts,
Peers: mp.Peers,
MissNodes: mp.MissNodes,
LoadResponse: mp.LoadResponse,
}
return mpInfo
}

sendOkReply(w, r, newSuccessHTTPReply(toInfo(mp)))
}

func parseAndExtractPartitionInfo(r *http.Request) (partitionID uint64, err error) {
Expand Down
15 changes: 8 additions & 7 deletions master/api_service_test.go
View file
Original file line number Diff line number Diff line change
Expand Up @@ -19,17 +19,18 @@ import (
"encoding/hex"
"encoding/json"
"fmt"
"github.com/chubaofs/chubaofs/master/mocktest"
"github.com/chubaofs/chubaofs/proto"
"github.com/chubaofs/chubaofs/util/config"
"github.com/chubaofs/chubaofs/util/log"
"io/ioutil"
"net/http"
_ "net/http/pprof"
"os"
"strings"
"testing"
"time"

"github.com/chubaofs/chubaofs/master/mocktest"
"github.com/chubaofs/chubaofs/proto"
"github.com/chubaofs/chubaofs/util/config"
"github.com/chubaofs/chubaofs/util/log"
)

const (
Expand Down Expand Up @@ -65,10 +66,10 @@ func createDefaultMasterServerForTest() *Server {
"retainLogs":"20000",
"tickInterval":500,
"electionTick":6,
"logDir": "/export/chubaofs/Logs",
"logDir": "/tmp/chubaofs/Logs",
"logLevel":"DEBUG",
"walDir":"/export/chubaofs/raft",
"storeDir":"/export/chubaofs/rocksdbstore",
"walDir":"/tmp/chubaofs/raft",
"storeDir":"/tmp/chubaofs/rocksdbstore",
"clusterName":"chubaofs"
}`
testServer, err := createMasterServer(cfgJSON)
Expand Down
17 changes: 9 additions & 8 deletions master/cluster.go
View file
Original file line number Diff line number Diff line change
Expand Up @@ -16,13 +16,14 @@ package master

import (
"fmt"
"sync"
"time"

"github.com/chubaofs/chubaofs/proto"
"github.com/chubaofs/chubaofs/raftstore"
"github.com/chubaofs/chubaofs/util"
"github.com/chubaofs/chubaofs/util/errors"
"github.com/chubaofs/chubaofs/util/log"
"sync"
"time"
)

// Cluster stores all the cluster-level information.
Expand Down Expand Up @@ -1268,21 +1269,21 @@ func (c *Cluster) metaNodeCount() (len int) {
return
}

func (c *Cluster) allDataNodes() (dataNodes []NodeView) {
dataNodes = make([]NodeView, 0)
func (c *Cluster) allDataNodes() (dataNodes []proto.NodeView) {
dataNodes = make([]proto.NodeView, 0)
c.dataNodes.Range(func(addr, node interface{}) bool {
dataNode := node.(*DataNode)
dataNodes = append(dataNodes, NodeView{Addr: dataNode.Addr, Status: dataNode.isActive, ID: dataNode.ID, IsWritable: dataNode.isWriteAble()})
dataNodes = append(dataNodes, proto.NodeView{Addr: dataNode.Addr, Status: dataNode.isActive, ID: dataNode.ID, IsWritable: dataNode.isWriteAble()})
return true
})
return
}

func (c *Cluster) allMetaNodes() (metaNodes []NodeView) {
metaNodes = make([]NodeView, 0)
func (c *Cluster) allMetaNodes() (metaNodes []proto.NodeView) {
metaNodes = make([]proto.NodeView, 0)
c.metaNodes.Range(func(addr, node interface{}) bool {
metaNode := node.(*MetaNode)
metaNodes = append(metaNodes, NodeView{ID: metaNode.ID, Addr: metaNode.Addr, Status: metaNode.IsActive, IsWritable: metaNode.isWritable()})
metaNodes = append(metaNodes, proto.NodeView{ID: metaNode.ID, Addr: metaNode.Addr, Status: metaNode.IsActive, IsWritable: metaNode.isWritable()})
return true
})
return
Expand Down
24 changes: 8 additions & 16 deletions master/cluster_stat.go
View file
Original file line number Diff line number Diff line change
Expand Up @@ -16,30 +16,22 @@ package master

import (
"fmt"
"strconv"

"github.com/chubaofs/chubaofs/proto"
"github.com/chubaofs/chubaofs/util"
"github.com/chubaofs/chubaofs/util/log"
"strconv"
)

type nodeStatInfo struct {
TotalGB uint64
UsedGB uint64
IncreasedGB int64
UsedRatio string
}
type nodeStatInfo = proto.NodeStatInfo

type volStatInfo struct {
Name string
TotalGB uint64
UsedGB uint64
UsedRatio string
}
type volStatInfo = proto.VolStatInfo

func newVolStatInfo(name string, total, used uint64, ratio string) *volStatInfo {
return &volStatInfo{
Name: name,
TotalGB: total,
UsedGB: used,
TotalSize: total,
UsedSize: used,
UsedRatio: ratio,
}
}
Expand Down Expand Up @@ -118,6 +110,6 @@ func (c *Cluster) updateVolStatInfo() {
continue
}
useRate := float64(used) / float64(total)
c.volStatInfo.Store(vol.Name, newVolStatInfo(vol.Name, total/util.GB, used/util.GB, strconv.FormatFloat(useRate, 'f', 3, 32)))
c.volStatInfo.Store(vol.Name, newVolStatInfo(vol.Name, total, used, strconv.FormatFloat(useRate, 'f', 3, 32)))
}
}
Loading

0 comments on commit 90daee7

Please sign in to comment.