gian/cubefs
1
0
Fork 0
mirror of https://github.com/cubefs/cubefs.git synced 2026-08-02 02:00:56 +00:00
Code Issues Actions Packages Projects Releases Wiki Activity

feat(master): add http interface getOpLog for data processing #22257509

Browse Source 
Signed-off-by: Reliey <zhumingze@oppo.com>
This commit is contained in:
Reliey 2024-10-24 15:39:01 +08:00 committed by AmazingChi
parent 5f3c6e2d4a
commit ddfc4c29b8
11 changed files with 551 additions and 191 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

98
cli/cmd/cluster.go
View File

@ -48,6 +48,8 @@ func newClusterCmd(client *master.MasterClient) *cobra.Command {
newClusterQueryDecommissionFailedDiskCmd(client),
// newClusterSetDecommissionDiskLimitCmd(client),
newClusterQueryDataNodeOpCmd(client),
newClusterQueryDpOpCmd(client),
newClusterQueryDiskOpCmd(client),
)
return clusterCmd
}
@ -72,12 +74,12 @@ const (
cmdQueryDecommissionFailedDiskShort = "query auto or manual decommission failed disk"
// cmdSetDecommissionDiskLimit = "set decommission disk limit"
cmdQueryDataNodeOpShort = "query DataNode_op information of a cluster"
cmdQueryDpOpShort = "query Dp_op information of a cluster"
cmdQueryDiskOpShort = "query Disk_op information of a cluster"
)
func newClusterInfoCmd(client *master.MasterClient) *cobra.Command {
var volStorageClass bool
var statOpLog bool
cmd := &cobra.Command{
Use: CliOpInfo,
Short: cmdClusterInfoShort,
@ -87,7 +89,7 @@ func newClusterInfoCmd(client *master.MasterClient) *cobra.Command {
var cn *proto.ClusterNodeInfo
var cp *proto.ClusterIP
var clusterPara map[string]string
if cv, err = client.AdminAPI().GetCluster(volStorageClass, statOpLog); err != nil {
if cv, err = client.AdminAPI().GetCluster(volStorageClass); err != nil {
errout(err)
}
if cn, err = client.AdminAPI().GetClusterNodeInfo(); err != nil {
@ -546,28 +548,100 @@ func newClusterQueryDecommissionFailedDiskCmd(client *master.MasterClient) *cobr
}
func newClusterQueryDataNodeOpCmd(client *master.MasterClient) *cobra.Command {
var filterOp string
var dataNodeName string
var logNum int
var volStorageClass bool
var statOpLog bool
var (
filterOp string
dataNodeName string
logNum int
dimension string
volName string
addr string
dpId string
diskName string
)
cmd := &cobra.Command{
Use: CliOpDataNodeOp,
Short: cmdQueryDataNodeOpShort,
RunE: func(cmd *cobra.Command, args []string) error {
statOpLog = true
cv, err := client.AdminAPI().GetCluster(volStorageClass, statOpLog)
dimension = proto.Node
opv, err := client.AdminAPI().GetOpLog(dimension, volName, addr, dpId, diskName)
if err != nil {
return err
}
stdoutln(fmt.Sprintf("%-30v %-20v %v", "Ip", "OpType", "Count"))
stdoutln(formatDataNodeOp(cv, logNum, dataNodeName, filterOp))
stdoutln(formatDataNodeOp(opv, logNum, dataNodeName, filterOp))
return nil
},
}
cmd.Flags().IntVar(&logNum, "num", 50, "Number of logs to display")
cmd.Flags().StringVar(&dataNodeName, "dataNode", "", "Filter logs by dataNode name")
cmd.Flags().StringVar(&filterOp, "filter-op", "", "Filter operations by type")
cmd.Flags().StringVar(&filterOp, "filter-op", "", "Filter logs by op type")
cmd.Flags().StringVar(&addr, "addr", "", "Filter logs by data node address")
return cmd
}
func newClusterQueryDpOpCmd(client *master.MasterClient) *cobra.Command {
var (
filterOp string
dpId string
logNum int
dimension string
volName string
addr string
diskName string
)
cmd := &cobra.Command{
Use: CliOpDpOp,
Short: cmdQueryDpOpShort,
RunE: func(cmd *cobra.Command, args []string) error {
dimension = proto.Dp
opv, err := client.AdminAPI().GetOpLog(dimension, volName, addr, dpId, diskName)
if err != nil {
return err
}
stdoutln(fmt.Sprintf("%-30v %-20v %v", "DpId", "OpType", "Count"))
stdoutln(formatClusterDpOp(opv, logNum, filterOp))
return nil
},
}
cmd.Flags().IntVar(&logNum, "num", 50, "Number of logs to display")
cmd.Flags().StringVar(&dpId, "dp", "", "Filter logs by dp id")
cmd.Flags().StringVar(&filterOp, "filter-op", "", "Filter logs by op type")
cmd.Flags().StringVar(&addr, "addr", "", "Filter logs by data node address")
return cmd
}
func newClusterQueryDiskOpCmd(client *master.MasterClient) *cobra.Command {
var (
filterOp string
diskName string
logNum int
dimension string
volName string
addr string
dpId string
)
cmd := &cobra.Command{
Use: CliOpDiskOp,
Short: cmdQueryDiskOpShort,
RunE: func(cmd *cobra.Command, args []string) error {
dimension = proto.Disk
opv, err := client.AdminAPI().GetOpLog(dimension, volName, addr, dpId, diskName)
if err != nil {
return err
}
if diskName == "" {
stdoutln(fmt.Sprintf("%-45v %-20v %v", "DiskName", "OpType", "Count"))
} else {
stdoutln(fmt.Sprintf("%-45v %-20v %v", "DpId", "OpType", "Count"))
}
stdoutln(formatClusterDiskOp(opv, logNum, filterOp))
return nil
},
}
cmd.Flags().IntVar(&logNum, "num", 50, "Number of logs to display")
cmd.Flags().StringVar(&diskName, "disk", "", "Filter logs by disk name")
cmd.Flags().StringVar(&filterOp, "filter-op", "", "Filter logs by op type")
cmd.Flags().StringVar(&addr, "addr", "", "Filter logs by data node address")
return cmd
}

124
cli/cmd/datanode.go
View File

@ -41,8 +41,8 @@ func newDataNodeCmd(client *master.MasterClient) *cobra.Command {
newDataNodeMigrateCmd(client),
newDataNodeQueryDecommissionedDisk(client),
newDataNodeCancelDecommissionCmd(client),
newDataNodeDiskOpCmd(client),
newDataNodeDpOpCmd(client),
// newDataNodeDiskOpCmd(client),
// newDataNodeDpOpCmd(client),
)
return cmd
}
@ -53,8 +53,8 @@ const (
cmdDataNodeDecommissionInfoShort = "decommission partitions in a data node to others"
cmdDataNodeQueryDecommissionedDisksShort = "query datanode decommissioned disks"
cmdDataNodeCancelDecommissionedDisksShort = "cancel decommission progress for datanode"
cmdDataNodeDiskOpShort = "Show Disk_op information of a data node"
cmdDataNodeDpOpShort = "Show Dp_op information of a data node"
// cmdDataNodeDiskOpShort = "Show Disk_op information of a data node"
// cmdDataNodeDpOpShort = "Show Dp_op information of a data node"
)
func newDataNodeListCmd(client *master.MasterClient) *cobra.Command {
@ -222,62 +222,62 @@ func newDataNodeCancelDecommissionCmd(client *master.MasterClient) *cobra.Comman
return cmd
}
func newDataNodeDiskOpCmd(client *master.MasterClient) *cobra.Command {
var filterOp string
var diskName string
var logNum int
cmd := &cobra.Command{
Use: CliOpDiskOp + " [{HOST}:{PORT}]",
Short: cmdDataNodeDiskOpShort,
Args: cobra.MinimumNArgs(1),
RunE: func(cmd *cobra.Command, args []string) error {
datanodeInfo, err := client.NodeAPI().GetDataNode(args[0])
if err != nil {
return err
}
stdoutln(fmt.Sprintf("%-30v %-20v %v", "DiskName", "OpType", "Count"))
stdoutln(formatDataNodeDiskOp(datanodeInfo, logNum, diskName, filterOp))
return nil
},
ValidArgsFunction: func(cmd *cobra.Command, args []string, toComplete string) ([]string, cobra.ShellCompDirective) {
if len(args) != 0 {
return nil, cobra.ShellCompDirectiveNoFileComp
}
return validDataNodes(client, toComplete), cobra.ShellCompDirectiveNoFileComp
},
}
cmd.Flags().IntVar(&logNum, "num", 50, "Number of logs to display")
cmd.Flags().StringVar(&diskName, "disk", "", "Filter logs by disk name")
cmd.Flags().StringVar(&filterOp, "filter-op", "", "Filter operations by type")
return cmd
}
// func newDataNodeDiskOpCmd(client *master.MasterClient) *cobra.Command {
// var filterOp string
// var diskName string
// var logNum int
// cmd := &cobra.Command{
// Use: CliOpDiskOp + " [{HOST}:{PORT}]",
// Short: cmdDataNodeDiskOpShort,
// Args: cobra.MinimumNArgs(1),
// RunE: func(cmd *cobra.Command, args []string) error {
// datanodeInfo, err := client.NodeAPI().GetDataNode(args[0])
// if err != nil {
// return err
// }
// stdoutln(fmt.Sprintf("%-30v %-20v %v", "DiskName", "OpType", "Count"))
// stdoutln(formatDataNodeDiskOp(datanodeInfo, logNum, diskName, filterOp))
// return nil
// },
// ValidArgsFunction: func(cmd *cobra.Command, args []string, toComplete string) ([]string, cobra.ShellCompDirective) {
// if len(args) != 0 {
// return nil, cobra.ShellCompDirectiveNoFileComp
// }
// return validDataNodes(client, toComplete), cobra.ShellCompDirectiveNoFileComp
// },
// }
// cmd.Flags().IntVar(&logNum, "num", 50, "Number of logs to display")
// cmd.Flags().StringVar(&diskName, "disk", "", "Filter logs by disk name")
// cmd.Flags().StringVar(&filterOp, "filter-op", "", "Filter operations by type")
// return cmd
// }
func newDataNodeDpOpCmd(client *master.MasterClient) *cobra.Command {
var filterOp string
var dpId string
var logNum int
cmd := &cobra.Command{
Use: CliOpDpOp + " [{HOST}:{PORT}]",
Short: cmdDataNodeDpOpShort,
Args: cobra.MinimumNArgs(1),
RunE: func(cmd *cobra.Command, args []string) error {
datanodeInfo, err := client.NodeAPI().GetDataNode(args[0])
if err != nil {
return err
}
stdoutln(fmt.Sprintf("%-30v %-20v %v", "DpId", "OpType", "Count"))
stdoutln(formatDataNodeDpOp(datanodeInfo, logNum, dpId, filterOp))
return nil
},
ValidArgsFunction: func(cmd *cobra.Command, args []string, toComplete string) ([]string, cobra.ShellCompDirective) {
if len(args) != 0 {
return nil, cobra.ShellCompDirectiveNoFileComp
}
return validDataNodes(client, toComplete), cobra.ShellCompDirectiveNoFileComp
},
}
cmd.Flags().IntVar(&logNum, "num", 50, "Number of logs to display")
cmd.Flags().StringVar(&dpId, "dp", "", "Filter logs by dp id")
cmd.Flags().StringVar(&filterOp, "filter-op", "", "Filter operations by type")
return cmd
}
// func newDataNodeDpOpCmd(client *master.MasterClient) *cobra.Command {
// var filterOp string
// var dpId string
// var logNum int
// cmd := &cobra.Command{
// Use: CliOpDpOp + " [{HOST}:{PORT}]",
// Short: cmdDataNodeDpOpShort,
// Args: cobra.MinimumNArgs(1),
// RunE: func(cmd *cobra.Command, args []string) error {
// datanodeInfo, err := client.NodeAPI().GetDataNode(args[0])
// if err != nil {
// return err
// }
// stdoutln(fmt.Sprintf("%-30v %-20v %v", "DpId", "OpType", "Count"))
// stdoutln(formatDataNodeDpOp(datanodeInfo, logNum, dpId, filterOp))
// return nil
// },
// ValidArgsFunction: func(cmd *cobra.Command, args []string, toComplete string) ([]string, cobra.ShellCompDirective) {
// if len(args) != 0 {
// return nil, cobra.ShellCompDirectiveNoFileComp
// }
// return validDataNodes(client, toComplete), cobra.ShellCompDirectiveNoFileComp
// },
// }
// cmd.Flags().IntVar(&logNum, "num", 50, "Number of logs to display")
// cmd.Flags().StringVar(&dpId, "dp", "", "Filter logs by dp id")
// cmd.Flags().StringVar(&filterOp, "filter-op", "", "Filter operations by type")
// return cmd
// }

151
cli/cmd/fmt.go
View File

@ -116,17 +116,14 @@ func formatClusterStat(cs *proto.ClusterStatInfo) string {
return sb.String()
}
func formatDataNodeOp(cv *proto.ClusterView, logNum int, dataNodeName string, filterOp string) string {
func formatDataNodeOp(opv *proto.OpLogView, logNum int, dataNodeName string, filterOp string) string {
maxLines := 1000
if logNum > 0 && logNum < maxLines {
maxLines = logNum
}
sb := strings.Builder{}
sort.Slice(cv.ClusterOpLogs, func(i, j int) bool {
return cv.ClusterOpLogs[i].Count > cv.ClusterOpLogs[j].Count
})
lineCount := 0
for _, opLog := range cv.ClusterOpLogs {
for _, opLog := range opv.ClusterOpLogs {
if dataNodeName != "" && opLog.Name != dataNodeName {
continue
}
@ -142,6 +139,46 @@ func formatDataNodeOp(cv *proto.ClusterView, logNum int, dataNodeName string, fi
return sb.String()
}
func formatClusterDpOp(opv *proto.OpLogView, logNum int, filterOp string) string {
maxLines := 1000
if logNum > 0 && logNum < maxLines {
maxLines = logNum
}
sb := strings.Builder{}
lineCount := 0
for _, opLog := range opv.DpOpLogs {
if filterOp != "" && !strings.Contains(opLog.Op, filterOp) {
continue
}
lineCount++
if lineCount > maxLines {
break
}
sb.WriteString(fmt.Sprintf("%-30v %-20v %v\n", opLog.Name, opLog.Op, opLog.Count))
}
return sb.String()
}
func formatClusterDiskOp(opv *proto.OpLogView, logNum int, filterOp string) string {
maxLines := 1000
if logNum > 0 && logNum < maxLines {
maxLines = logNum
}
sb := strings.Builder{}
lineCount := 0
for _, opLog := range opv.DiskOpLogs {
if filterOp != "" && !strings.Contains(opLog.Op, filterOp) {
continue
}
lineCount++
if lineCount > maxLines {
break
}
sb.WriteString(fmt.Sprintf("%-45v %-20v %v\n", opLog.Name, opLog.Op, opLog.Count))
}
return sb.String()
}
var nodeViewTableRowPattern = "%-6v %-65v %-8v %-8v %-8v %-12v"
func formatNodeViewTableHeader() string {
@ -250,22 +287,16 @@ func formatSimpleVolView(svv *proto.SimpleVolView) string {
return sb.String()
}
func formatVolOp(cv *proto.ClusterView, logNum int, volName string, dpId string, filterOp string) string {
func formatVolOp(opv *proto.OpLogView, logNum int, dpId string, filterOp string) string {
maxLines := 1000
if logNum > 0 && logNum < maxLines {
maxLines = logNum
}
sb := strings.Builder{}
sort.Slice(cv.VolOpLogs, func(i, j int) bool {
return cv.VolOpLogs[i].Count > cv.VolOpLogs[j].Count
})
lineCount := 0
for _, opLog := range cv.VolOpLogs {
for _, opLog := range opv.VolOpLogs {
parts := strings.Split(opLog.Name, "_")
if volName != "" && parts[0] != volName {
continue
}
if dpId != "" && parts[1] != dpId {
if dpId != "" && parts[0] != dpId {
continue
}
if filterOp != "" && !strings.Contains(opLog.Op, filterOp) {
@ -275,7 +306,7 @@ func formatVolOp(cv *proto.ClusterView, logNum int, volName string, dpId string,
if lineCount > maxLines {
break
}
sb.WriteString(fmt.Sprintf("%-20v %-15v %-15v %v\n", parts[0], "dp_"+parts[1], opLog.Op, opLog.Count))
sb.WriteString(fmt.Sprintf("%-15v %-15v %v\n", "dp_"+parts[0], opLog.Op, opLog.Count))
}
return sb.String()
}
@ -931,52 +962,52 @@ func formatDataNodeDetail(dn *proto.DataNodeInfo, rowTable bool) string {
return sb.String()
}
func formatDataNodeDiskOp(dn *proto.DataNodeInfo, logNum int, diskName string, filterOp string) string {
maxLines := 1000
if logNum > 0 && logNum < maxLines {
maxLines = logNum
}
sb := strings.Builder{}
lineCount := 0
for _, opLog := range dn.DiskOpLogs {
if diskName != "" && opLog.Name != diskName {
continue
}
if filterOp != "" && !strings.Contains(opLog.Op, filterOp) {
continue
}
lineCount++
if lineCount > maxLines {
break
}
sb.WriteString(fmt.Sprintf("%-30v %-20v %v\n", opLog.Name, opLog.Op, opLog.Count))
}
return sb.String()
}
// func formatDataNodeDiskOp(dn *proto.DataNodeInfo, logNum int, diskName string, filterOp string) string {
// maxLines := 1000
// if logNum > 0 && logNum < maxLines {
// maxLines = logNum
// }
// sb := strings.Builder{}
// lineCount := 0
// for _, opLog := range dn.DiskOpLogs {
// if diskName != "" && opLog.Name != diskName {
// continue
// }
// if filterOp != "" && !strings.Contains(opLog.Op, filterOp) {
// continue
// }
// lineCount++
// if lineCount > maxLines {
// break
// }
// sb.WriteString(fmt.Sprintf("%-30v %-20v %v\n", opLog.Name, opLog.Op, opLog.Count))
// }
// return sb.String()
// }
func formatDataNodeDpOp(dn *proto.DataNodeInfo, logNum int, dpId string, filterOp string) string {
maxLines := 1000
if logNum > 0 && logNum < maxLines {
maxLines = logNum
}
sb := strings.Builder{}
lineCount := 0
for _, opLog := range dn.DpOpLogs {
arr := strings.Split(opLog.Name, "_")
if dpId != "" && arr[1] != dpId {
continue
}
if filterOp != "" && !strings.Contains(opLog.Op, filterOp) {
continue
}
lineCount++
if lineCount > maxLines {
break
}
sb.WriteString(fmt.Sprintf("%-30v %-20v %v\n", opLog.Name, opLog.Op, opLog.Count))
}
return sb.String()
}
// func formatDataNodeDpOp(dn *proto.DataNodeInfo, logNum int, dpId string, filterOp string) string {
// maxLines := 1000
// if logNum > 0 && logNum < maxLines {
// maxLines = logNum
// }
// sb := strings.Builder{}
// lineCount := 0
// for _, opLog := range dn.DpOpLogs {
// arr := strings.Split(opLog.Name, "_")
// if dpId != "" && arr[1] != dpId {
// continue
// }
// if filterOp != "" && !strings.Contains(opLog.Op, filterOp) {
// continue
// }
// lineCount++
// if lineCount > maxLines {
// break
// }
// sb.WriteString(fmt.Sprintf("%-30v %-20v %v\n", opLog.Name, opLog.Op, opLog.Count))
// }
// return sb.String()
// }
var metaNodeDetailTableRowPattern = "%-6v %-6v %-65v %-6v %-6v %-6v %-10v"

28
cli/cmd/vol.go
View File

@ -1391,30 +1391,34 @@ func newVolAddAllowedStorageClassCmd(client *master.MasterClient) *cobra.Command
}
func newVolQueryOpCmd(client *master.MasterClient) *cobra.Command {
var filterOp string
var dpId string
var volName string
var logNum int
var volStorageClass bool
var statOpLog bool
var (
filterOp string
dpId string
// volName string
logNum int
dimension string
addr string
diskName string
)
cmd := &cobra.Command{
Use: CliOpVolOp,
Short: cmdQueryOpShort,
Args: cobra.MinimumNArgs(1),
RunE: func(cmd *cobra.Command, args []string) error {
statOpLog = true
cv, err := client.AdminAPI().GetCluster(volStorageClass, statOpLog)
dimension = proto.Vol
opv, err := client.AdminAPI().GetOpLog(dimension, args[0], addr, dpId, diskName)
if err != nil {
return err
}
stdoutln(fmt.Sprintf("%-20v %-15v %-15v %v", "VolName", "DpId", "OpType", "Count"))
stdoutln(formatVolOp(cv, logNum, volName, dpId, filterOp))
stdoutln(fmt.Sprintf("%-15v %-15v %v", "DpId", "OpType", "Count"))
stdoutln(formatVolOp(opv, logNum, dpId, filterOp))
return nil
},
}
cmd.Flags().IntVar(&logNum, "num", 50, "Number of logs to display")
cmd.Flags().StringVar(&volName, "volname", "", "Filter logs by vol name")
// cmd.Flags().StringVar(&volName, "volname", "", "Filter logs by vol name")
cmd.Flags().StringVar(&dpId, "dp", "", "Filter logs by dp id")
cmd.Flags().StringVar(&filterOp, "filter-op", "", "Filter operations by type")
cmd.Flags().StringVar(&filterOp, "filter-op", "", "Filter logs by op type")
return cmd
}

91
master/api_service.go
View File

@ -779,9 +779,68 @@ func (m *Server) aclOperate(w http.ResponseWriter, r *http.Request) {
_ = sendOkReply(w, r, newSuccessHTTPReply(rsp))
}
func (m *Server) getOpLog(w http.ResponseWriter, r *http.Request) {
var (
dimension string
volName string
addr string
dpId string
diskName string
)
metric := exporter.NewTPCnt(apiToMetricsName(proto.AdminGetOpLog))
defer func() {
doStatAndMetric(proto.AdminGetOpLog, metric, nil, nil)
}()
if err := r.ParseForm(); err != nil {
sendErrReply(w, r, newErrHTTPReply(err))
return
}
dimension = r.FormValue(opLogDimensionKey)
volName = r.FormValue(volNameKey)
addr = r.FormValue(addrKey)
dpId = r.FormValue(dpIdKey)
diskName = r.FormValue(diskNameKey)
var opv proto.OpLogView
switch dimension {
case proto.Dp:
opv = m.cluster.getDpOpLog(addr, dpId)
sort.Slice(opv.DpOpLogs, func(i, j int) bool {
return opv.DpOpLogs[i].Count > opv.DpOpLogs[j].Count
})
case proto.Disk:
opv = m.cluster.getDiskOpLog(addr, diskName)
sort.Slice(opv.DiskOpLogs, func(i, j int) bool {
return opv.DiskOpLogs[i].Count > opv.DiskOpLogs[j].Count
})
case proto.Node:
opv = m.cluster.getClusterOpLog()
sort.Slice(opv.ClusterOpLogs, func(i, j int) bool {
return opv.ClusterOpLogs[i].Count > opv.ClusterOpLogs[j].Count
})
case proto.Vol:
opv = m.cluster.getVolOpLog(volName)
sort.Slice(opv.VolOpLogs, func(i, j int) bool {
return opv.VolOpLogs[i].Count > opv.VolOpLogs[j].Count
})
default:
sendErrReply(w, r, newErrHTTPReply(fmt.Errorf("Invalid dimension: %s", dimension)))
return
}
sendOkReply(w, r, newSuccessHTTPReply(opv))
}
func (m *Server) getCluster(w http.ResponseWriter, r *http.Request) {
var volStorageClass bool
var statOpLog bool
metric := exporter.NewTPCnt(apiToMetricsName(proto.AdminGetCluster))
defer func() {
@ -802,15 +861,6 @@ func (m *Server) getCluster(w http.ResponseWriter, r *http.Request) {
}
}
if value := r.FormValue(statOpLogKey); value != "" {
var err error
statOpLog, err = strconv.ParseBool(value)
if err != nil {
sendErrReply(w, r, newErrHTTPReply(err))
return
}
}
cv := &proto.ClusterView{
Name: m.cluster.Name,
CreateTime: time.Unix(m.cluster.CreateTime, 0).Format(proto.TimeFormat),
@ -922,27 +972,6 @@ func (m *Server) getCluster(w http.ResponseWriter, r *http.Request) {
}
if statOpLog {
var volUpdated bool
for _, dv := range cv.DataNodes {
value, ok := m.cluster.dataNodes.Load(dv.Addr)
if !ok {
log.LogErrorf("data node %s is not exist", dv.Addr)
continue
}
dataNode := value.(*DataNode)
dataNodeOpLogs := make([]proto.OpLog, 0)
dataNode.updateDataNodeOpLog(&dataNodeOpLogs)
cv.ClusterOpLogs = append(cv.ClusterOpLogs, dataNodeOpLogs...)
if !volUpdated {
dataNode.updateVolOpLog(m.cluster, &cv.VolOpLogs)
volUpdated = true
}
}
}
cv.BadPartitionIDs = m.cluster.getBadDataPartitionsView()
cv.BadMetaPartitionIDs = m.cluster.getBadMetaPartitionsView()

181
master/cluster.go
View File

@ -6101,3 +6101,184 @@ func (c *Cluster) processDataPartitionDecommission(id uint64) bool {
}
return false
}
func (c *Cluster) getDpOpLog(addr string, dpId string) proto.OpLogView {
var opv proto.OpLogView
opCounts := make(map[string]int32)
if addr != "" {
dataNode, err := c.dataNode(addr)
if err != nil {
log.LogErrorf("get dataNode failed, err(%v)", err.Error())
return opv
}
for _, opLog := range dataNode.DpOpLogs {
parts := strings.Split(opLog.Name, "_")
if len(parts) < 3 {
log.LogErrorf("Invalid opLog name format: %s", opLog.Name)
continue
}
opv.DpOpLogs = append(opv.DpOpLogs, proto.OpLog{
Name: parts[0] + "_" + parts[1],
Op: parts[2],
Count: opLog.Count,
})
}
return opv
}
if dpId != "" {
id, err := strconv.ParseUint(dpId, 10, 64)
if err != nil {
log.LogErrorf("failed to transform dpId, err(%v)", err)
return opv
}
dp, err := c.getDataPartitionByID(id)
if err != nil {
log.LogErrorf("failed to get dp(%v), err(%v)", dpId, err)
return opv
}
for _, host := range dp.Hosts {
dataNode, err := c.dataNode(host)
if err != nil {
log.LogErrorf("get dataNode failed, err(%v)", err.Error())
return opv
}
for _, opLog := range dataNode.DpOpLogs {
parts := strings.Split(opLog.Name, "_")
if len(parts) < 3 {
log.LogErrorf("Invalid opLog name format: %s", opLog.Name)
continue
}
if parts[1] == dpId {
opv.DpOpLogs = append(opv.DpOpLogs, proto.OpLog{
Name: fmt.Sprintf("%s [%s]", parts[0]+"_"+parts[1], dataNode.Addr),
Op: parts[2],
Count: opLog.Count,
})
}
}
}
return opv
}
c.dataNodes.Range(func(key, node interface{}) bool {
dataNode := node.(*DataNode)
for _, opLog := range dataNode.DpOpLogs {
if curCount, exists := opCounts[opLog.Name]; !exists || curCount < opLog.Count {
opCounts[opLog.Name] = opLog.Count
}
}
return true
})
for key, count := range opCounts {
parts := strings.Split(key, "_")
if len(parts) < 3 {
log.LogErrorf("Invalid opLog name format: %s", key)
continue
}
opv.DpOpLogs = append(opv.DpOpLogs, proto.OpLog{
Name: parts[0] + "_" + parts[1],
Op: parts[2],
Count: count,
})
}
return opv
}
func (c *Cluster) getDiskOpLog(addr string, diskName string) proto.OpLogView {
var opv proto.OpLogView
if addr != "" && diskName != "" {
dataNode, err := c.dataNode(addr)
if err != nil {
log.LogErrorf("get dataNode failed, err(%v)", err.Error())
return opv
}
for _, opLog := range dataNode.DpOpLogs {
parts := strings.Split(opLog.Name, "_")
if len(parts) < 3 {
log.LogErrorf("Invalid opLog name format: %s", opLog.Name)
continue
}
dpId, err := strconv.ParseUint(parts[1], 10, 64)
if err != nil {
log.LogErrorf("failed to transform dpId, err(%v)", err)
return opv
}
dp, err := c.getDataPartitionByID(dpId)
if err != nil {
log.LogErrorf("failed to get dp(%v), err(%v)", dpId, err)
return opv
}
for _, replica := range dp.Replicas {
if replica.Addr != addr || replica.DiskPath != diskName {
continue
}
opv.DiskOpLogs = append(opv.DiskOpLogs, proto.OpLog{
Name: opLog.Name,
Op: opLog.Op,
Count: opLog.Count,
})
}
}
return opv
}
if addr != "" {
dataNode, err := c.dataNode(addr)
if err != nil {
log.LogErrorf("get dataNode failed, err(%v)", err.Error())
return opv
}
opv.DiskOpLogs = append(opv.DiskOpLogs, dataNode.DiskOpLogs...)
return opv
}
c.dataNodes.Range(func(key, node interface{}) bool {
dataNode := node.(*DataNode)
for _, opLog := range dataNode.DiskOpLogs {
opv.DiskOpLogs = append(opv.DiskOpLogs, proto.OpLog{
Name: fmt.Sprintf("%s [%s]", opLog.Name, dataNode.Addr),
Op: opLog.Op,
Count: opLog.Count,
})
}
return true
})
return opv
}
func (c *Cluster) getClusterOpLog() proto.OpLogView {
var opv proto.OpLogView
c.dataNodes.Range(func(addr, node interface{}) bool {
dataNode := node.(*DataNode)
dataNodeOpLogs := dataNode.getDataNodeOpLog()
opv.ClusterOpLogs = append(opv.ClusterOpLogs, dataNodeOpLogs...)
return true
})
return opv
}
func (c *Cluster) getVolOpLog(volName string) proto.OpLogView {
var opv proto.OpLogView
opCounts := make(map[string]int32)
c.dataNodes.Range(func(addr, node interface{}) bool {
dataNode := node.(*DataNode)
volOpLogs := dataNode.getVolOpLog(c, volName)
for _, opLog := range volOpLogs {
newName := opLog.Name + "_" + opLog.Op
if curCount, exists := opCounts[newName]; !exists || curCount < opLog.Count {
opCounts[newName] = opLog.Count
}
}
return true
})
for key, count := range opCounts {
parts := strings.Split(key, "_")
if len(parts) < 3 {
log.LogErrorf("Invalid opLog name format: %s", key)
continue
}
opv.VolOpLogs = append(opv.VolOpLogs, proto.OpLog{
Name: parts[1],
Op: parts[2],
Count: count,
})
}
return opv
}

5
master/const.go
View File

@ -154,7 +154,10 @@ const (
mediaTypeKey = "mediaType"
allowedStorageClassKey = "allowedStorageClass"
volStorageClassKey = "volStorageClass"
statOpLogKey = "statOpLog"
opLogDimensionKey = "opLogDimension"
volNameKey = "volName"
dpIdKey = "dpId"
diskNameKey = "diskName"
forbidWriteOpOfProtoVersion0 = "forbidWriteOpOfProtoVersion0"
quotaClass = "quotaClass"
quotaOfClass = "quotaOfStorageClass"

20
master/data_node.go
View File

@ -260,21 +260,25 @@ func (dataNode *DataNode) updateNodeMetric(c *Cluster, resp *proto.DataNodeHeart
dataNode.Total, dataNode.Used, dataNode.AvailableSpace)
}
func (dataNode *DataNode) updateDataNodeOpLog(dataNodeOpLogs *[]proto.OpLog) {
func (dataNode *DataNode) getDataNodeOpLog() []proto.OpLog {
dataNodeOpLogs := make([]proto.OpLog, 0)
opCounts := make(map[string]int32)
for _, opLog := range dataNode.DiskOpLogs {
opCounts[opLog.Op] += opLog.Count
}
for op, count := range opCounts {
*dataNodeOpLogs = append(*dataNodeOpLogs, proto.OpLog{
dataNodeOpLogs = append(dataNodeOpLogs, proto.OpLog{
Name: dataNode.Addr,
Op: op,
Count: count,
})
}
return dataNodeOpLogs
}
func (dataNode *DataNode) updateVolOpLog(c *Cluster, volOpLogs *[]proto.OpLog) {
func (dataNode *DataNode) getVolOpLog(c *Cluster, volName string) []proto.OpLog {
volOpLogs := make([]proto.OpLog, 0)
for _, opLog := range dataNode.DpOpLogs {
parts := strings.Split(opLog.Name, "_")
dpId, err := strconv.ParseUint(parts[1], 10, 64)
@ -289,14 +293,18 @@ func (dataNode *DataNode) updateVolOpLog(c *Cluster, volOpLogs *[]proto.OpLog) {
continue
}
volName := dp.VolName
newName := fmt.Sprintf("%s_%d", volName, dpId)
*volOpLogs = append(*volOpLogs, proto.OpLog{
if dp.VolName != volName {
continue
}
newName := fmt.Sprintf("%s_%d", dp.VolName, dpId)
volOpLogs = append(volOpLogs, proto.OpLog{
Name: newName,
Op: opLog.Op,
Count: opLog.Count,
})
}
return volOpLogs
}
func (dataNode *DataNode) canAlloc() bool {

3
master/http_server.go
View File

@ -284,6 +284,9 @@ func (m *Server) registerAPIRoutes(router *mux.Router) {
router.NewRoute().Methods(http.MethodGet).
Path(proto.AdminGetCluster).
HandlerFunc(m.getCluster)
router.NewRoute().Methods(http.MethodGet).
Path(proto.AdminGetOpLog).
HandlerFunc(m.getOpLog)
router.NewRoute().Name(proto.AdminACL).
Methods(http.MethodGet).
Path(proto.AdminACL).

21
proto/admin_proto.go
View File

@ -93,6 +93,7 @@ const (
AdminQueryDecommissionFailedDisk = "/admin/queryDecommissionFailedDisk"
AdminAbortDecommissionDisk = "/admin/abortDecommissionDisk"
AdminResetDataPartitionRestoreStatus = "/admin/resetDataPartitionRestoreStatus"
AdminGetOpLog = "/admin/getOpLog"
// #nosec G101
AdminQueryDecommissionToken = "/admin/queryDecommissionToken"
@ -315,6 +316,7 @@ var GApiInfo map[string]string = map[string]string{
"admindatapartitionchangeleader": AdminDataPartitionChangeLeader,
"adminsetdpdiscard": AdminSetDpDiscard,
"admingetdiscarddp": AdminGetDiscardDp,
"admingetoplog": AdminGetOpLog,
// "adminclusterapi": AdminClusterAPI,
// "adminuserapi": AdminUserAPI,
@ -832,8 +834,8 @@ type DataNodeHeartbeatResponse struct {
CpuUtil float64 `json:"cpuUtil"`
IoUtils map[string]float64 `json:"ioUtil"`
BackupDataPartitions []BackupDataPartitionInfo
DiskOpLogs []OpLog
DpOpLogs []OpLog
DiskOpLogs []OpLog `json:"DiskOpLog"`
DpOpLogs []OpLog `json:"DpOpLog"`
ReceivedForbidWriteOpOfProtoVer0 bool
}
@ -1312,6 +1314,21 @@ type TopologyView struct {
Zones []*ZoneView
}
const (
Dp = "dpop"
Disk = "diskop"
Node = "datanodeop"
Vol = "volop"
)
// OpLogView defines the view of all opLogs
type OpLogView struct {
DpOpLogs []OpLog
DiskOpLogs []OpLog
ClusterOpLogs []OpLog
VolOpLogs []OpLog
}
const (
PartitionTypeNormal = 0
PartitionTypeCache = 1

20
sdk/master/api_admin.go
View File

@ -44,15 +44,25 @@ func (api *AdminAPI) EncodingGzip() *AdminAPI {
return api.EncodingWith(encodingGzip)
}
func (api *AdminAPI) GetCluster(volStorageClass bool, statOpLog bool) (cv *proto.ClusterView, err error) {
cv = &proto.ClusterView{}
err = api.mc.requestWith(cv, newRequest(get, proto.AdminGetCluster).Header(api.h).Param(
anyParam{"volStorageClass", strconv.FormatBool(volStorageClass)},
anyParam{"statOpLog", strconv.FormatBool(statOpLog)},
func (api *AdminAPI) GetOpLog(dimension string, volName string, addr string, dpId string, diskName string) (opv *proto.OpLogView, err error) {
opv = &proto.OpLogView{}
err = api.mc.requestWith(opv, newRequest(get, proto.AdminGetOpLog).Header(api.h).Param(
anyParam{"opLogDimension", dimension},
anyParam{"volName", volName},
anyParam{"addr", addr},
anyParam{"dpId", dpId},
anyParam{"diskName", diskName},
))
return
}
func (api *AdminAPI) GetCluster(volStorageClass bool) (cv *proto.ClusterView, err error) {
cv = &proto.ClusterView{}
err = api.mc.requestWith(cv, newRequest(get, proto.AdminGetCluster).Header(api.h).
addParam("volStorageClass", strconv.FormatBool(volStorageClass)))
return
}
func (api *AdminAPI) GetClusterDataNodes() (nodes []proto.NodeView, err error) {
nodes = []proto.NodeView{}
err = api.mc.requestWith(&nodes, newRequest(get, proto.AdminGetClusterDataNodes).Header(api.h))