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

refactor(all): resolve conflict when pick commit from 3.3.2 to 3.4.0. #22505175

Browse Source 
Signed-off-by: Victor1319 <zengxuewei@oppo.com>
This commit is contained in:
Victor1319 2024-08-15 20:19:14 +08:00
parent bd4cdbdd38
commit 83c50e1c50
17 changed files with 49 additions and 118 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

2
cli/cmd/vol.go
View File

@ -296,7 +296,7 @@ func newVolUpdateCmd(client *master.MasterClient) *cobra.Command {
var optTrashInterval int64
var optAccessTimeValidInterval int64
var optEnablePersistAccessTime string
var confirmString = strings.Builder{}
confirmString := strings.Builder{}
var vv *proto.SimpleVolView
cmd := &cobra.Command{
Use: CliOpUpdate + " [VOLUME NAME]",

3
datanode/space_manager.go
View File

@ -559,7 +559,8 @@ func (manager *SpaceManager) DeletePartition(dpID uint64, force bool) (err error
}
func (s *DataNode) buildHeartBeatResponse(response *proto.DataNodeHeartbeatResponse,
volNames map[string]struct{}, dpRepairBlockSize map[string]uint64, reqID string) {
volNames map[string]struct{}, dpRepairBlockSize map[string]uint64, reqID string,
) {
response.Status = proto.TaskSucceeds
stat := s.space.Stats()
stat.Lock()

18
datanode/wrap_operator.go
View File

@ -734,14 +734,18 @@ func (s *DataNode) handleBatchMarkDeletePacket(p *repl.Packet, c net.Conn) {
// if we prevent it, we will get "orphan extents"
var exts []*proto.DelExtentParam
err = json.Unmarshal(p.Data, &exts)
if err != nil {
log.LogErrorf("[handleBatchMarkDeletePacket] failed to unmarshal request, err(%v)", err)
return
}
store := partition.ExtentStore()
if err == nil {
for _, ext := range exts {
if p.Opcode == proto.OpGcBatchDeleteExtent && !store.CanGcDelete(ext.ExtentId) {
log.LogWarnf("handleBatchMarkDeletePacket: ext %d is not in gc status, can't be gc delete, dp %d", ext.ExtentId, ext.PartitionId)
err = storage.ParameterMismatchError
return
}
for _, ext := range exts {
if p.Opcode == proto.OpGcBatchDeleteExtent && !store.CanGcDelete(ext.ExtentId) {
log.LogWarnf("handleBatchMarkDeletePacket: ext %d is not in gc status, can't be gc delete, dp %d", ext.ExtentId, ext.PartitionId)
err = storage.ParameterMismatchError
return
}
if !deleteLimiteRater.Allow() {
log.LogInfof("[handleBatchMarkDeletePacket] delete limiter reach(%v), remote (%v) try again.", deleteLimiteRater.Limit(), c.RemoteAddr().String())

72
master/api_service.go
View File

@ -7414,75 +7414,3 @@ func (m *Server) resetDecommissionDataNodeStatus(w http.ResponseWriter, r *http.
msg := fmt.Sprintf("reset decommission status for datanode %v success", dn.Addr)
sendOkReply(w, r, newSuccessHTTPReply(msg))
}
func (m *Server) setVolAccessTimeValidInterval(w http.ResponseWriter, r *http.Request) {
var (
name string
interval int64
err error
msg string
authKey string
vol *Vol
)
metric := exporter.NewTPCnt(apiToMetricsName(proto.AdminSetVolAccessTimeValidInterval))
defer func() {
doStatAndMetric(proto.AdminSetVolAccessTimeValidInterval, metric, err, map[string]string{exporter.Vol: name})
}()
if name, authKey, interval, err = parseRequestToSetInterval(r); err != nil {
sendErrReply(w, r, &proto.HTTPReply{Code: proto.ErrCodeParamError, Msg: err.Error()})
return
}
if interval < 0 {
sendErrReply(w, r, &proto.HTTPReply{Code: proto.ErrCodeParamError, Msg: "interval cannot be less than 0"})
return
}
if vol, err = m.cluster.getVol(name); err != nil {
sendErrReply(w, r, &proto.HTTPReply{Code: proto.ErrCodeVolNotExists, Msg: err.Error()})
return
}
newArgs := getVolVarargs(vol)
newArgs.accessTimeInterval = interval
if err = m.cluster.updateVol(name, authKey, newArgs); err != nil {
sendErrReply(w, r, newErrHTTPReply(err))
return
}
msg = fmt.Sprintf("update vol[%v] accessTimePersistInterval to %v successfully\n", name, interval)
sendOkReply(w, r, newSuccessHTTPReply(msg))
}
func (m *Server) enablePersistVolAccessTime(w http.ResponseWriter, r *http.Request) {
var (
name string
enable bool
err error
msg string
authKey string
vol *Vol
)
metric := exporter.NewTPCnt(apiToMetricsName(proto.AdminSetVolAccessTimeValidInterval))
defer func() {
doStatAndMetric(proto.AdminSetVolAccessTimeValidInterval, metric, err, map[string]string{exporter.Vol: name})
}()
if name, authKey, enable, err = parseRequestToSetPersist(r); err != nil {
sendErrReply(w, r, &proto.HTTPReply{Code: proto.ErrCodeParamError, Msg: err.Error()})
return
}
if vol, err = m.cluster.getVol(name); err != nil {
sendErrReply(w, r, &proto.HTTPReply{Code: proto.ErrCodeVolNotExists, Msg: err.Error()})
return
}
newArgs := getVolVarargs(vol)
newArgs.enablePersistAccessTime = enable
if err = m.cluster.updateVol(name, authKey, newArgs); err != nil {
sendErrReply(w, r, newErrHTTPReply(err))
return
}
msg = fmt.Sprintf("update vol[%v] enablePersistAccessTime to %v successfully\n", name, enable)
sendOkReply(w, r, newSuccessHTTPReply(msg))
}

3
master/data_partition.go
View File

@ -1155,7 +1155,8 @@ func GetSpecialDecommissionStatusMessage(status uint32) string {
}
func (partition *DataPartition) MarkDecommissionStatus(srcAddr, dstAddr, srcDisk string, raftForce bool, term uint64,
migrateType uint32, c *Cluster, ns *nodeSet) (err error) {
migrateType uint32, c *Cluster, ns *nodeSet,
) (err error) {
defer func() {
if err != nil {
msg := fmt.Sprintf("dp(%v) mark decommission status failed", partition.decommissionInfo())

4
master/data_partition_check.go
View File

@ -20,8 +20,6 @@ import (
"strconv"
"time"
"github.com/cubefs/cubefs/datanode"
"github.com/cubefs/cubefs/proto"
"github.com/cubefs/cubefs/util"
"github.com/cubefs/cubefs/util/log"
@ -271,7 +269,7 @@ func (partition *DataPartition) checkMissingReplicas(clusterID, leaderAddr strin
if oldDpReplicaAliveNum != "" {
WarnMetrics.missingDp.DeleteLabelValues(clusterID, id, missingReplicaAddr, oldDpReplicaAliveNum, replicaInfo.replicaNum)
}
//WarnMetrics.missingDp.SetWithLabelValues(1, clusterID, id, missingReplicaAddr, replicaInfo.replicaAlive, replicaInfo.replicaNum)
// WarnMetrics.missingDp.SetWithLabelValues(1, clusterID, id, missingReplicaAddr, replicaInfo.replicaAlive, replicaInfo.replicaNum)
}
}
}

3
master/http_server.go
View File

@ -80,7 +80,6 @@ func (m *Server) isFollowerRead(r *http.Request) (followerRead bool) {
followerRead = false
if r.URL.Path == proto.ClientDataPartitions && !m.partition.IsRaftLeader() {
if volName, err := parseAndExtractName(r); err == nil {
log.LogInfof("action[interceptor] followerRead vol[%v]", volName)
if followerRead = m.cluster.followerReadManager.IsVolViewReady(volName); followerRead {
@ -874,13 +873,11 @@ func ErrResponse(w http.ResponseWriter, err error) {
}
func (m *Server) newReverseProxy() *httputil.ReverseProxy {
tr := &http.Transport{}
if m.config != nil {
tr = proto.GetHttpTransporter(&proto.HttpCfg{
PoolSize: int(m.config.httpProxyPoolSize),
})
}
return &httputil.ReverseProxy{

7
master/monitor_metrics.go
View File

@ -357,7 +357,7 @@ func (m *warningMetrics) WarnDpNoLeader(clusterName string, partitionID uint64,
}
if now-info.ReportTime > m.cluster.cfg.DpNoLeaderReportIntervalSec {
if m.dpNoLeader != nil {
//m.dpNoLeader.SetWithLabelValues(1, clusterName, strconv.FormatUint(partitionID, 10), strconv.FormatUint(uint64(replicas), 10))
// m.dpNoLeader.SetWithLabelValues(1, clusterName, strconv.FormatUint(partitionID, 10), strconv.FormatUint(uint64(replicas), 10))
}
m.dpNoLeaderInfo[partitionID] = NoLeaderPartInfo{ReportTime: now, Replicas: replicas}
}
@ -403,7 +403,7 @@ func (m *warningMetrics) WarnMissingMp(clusterName, addr string, partitionID uin
}
if m.missingMp != nil {
//m.missingMp.SetWithLabelValues(1, clusterName, id, addr)
// m.missingMp.SetWithLabelValues(1, clusterName, id, addr)
}
if _, ok := m.mpMissingReplicaInfo[id]; !ok {
m.mpMissingReplicaInfo[id] = addrSet{addrs: make(map[string]voidType)}
@ -462,7 +462,7 @@ func (m *warningMetrics) WarnMpNoLeader(clusterName string, partitionID uint64,
if now-info.ReportTime > m.cluster.cfg.MpNoLeaderReportIntervalSec {
if m.mpNoLeader != nil {
//m.mpNoLeader.SetWithLabelValues(1, clusterName, strconv.FormatUint(partitionID, 10), strconv.FormatUint(uint64(replicas), 10))
// m.mpNoLeader.SetWithLabelValues(1, clusterName, strconv.FormatUint(partitionID, 10), strconv.FormatUint(uint64(replicas), 10))
}
m.mpNoLeaderInfo[partitionID] = NoLeaderPartInfo{ReportTime: now, Replicas: replicas}
}
@ -590,7 +590,6 @@ func (mm *monitorMetrics) doStat() {
}
func (mm *monitorMetrics) setMpAndDpMetrics() {
start := time.Now()
defer func() {
log.LogInfof("setMpAndDpMetrics: total cost %d ms", time.Since(start).Milliseconds())

2
master/vol.go
View File

@ -224,7 +224,7 @@ func newVol(vv volValue) (vol *Vol) {
vol.preloadCapacity = math.MaxUint64 // mark as special value to trigger calculate
vol.dpRepairBlockSize = proto.DefaultDpRepairBlockSize
vol.EnableAutoMetaRepair.Store(defaultEnableDpMetaRepair)
vol.preloadCapacity = math.MaxUint64 //mark as special value to trigger calculate
vol.preloadCapacity = math.MaxUint64 // mark as special value to trigger calculate
vol.TrashInterval = vv.TrashInterval
vol.AccessTimeValidInterval = vv.AccessTimeInterval
vol.EnablePersistAccessTime = vv.EnablePersistAccessTime

1
metanode/api_handler.go
View File

@ -926,7 +926,6 @@ func (m *MetaNode) getInodeAccessTimeHandler(w http.ResponseWriter, r *http.Requ
defer func() {
data, _ := resp.Marshal()
if _, err := w.Write(data); err != nil {
log.LogErrorf("[getInodeHandler] response %s", err)
}
}()

3
metanode/manager_op.go
View File

@ -2767,7 +2767,8 @@ end:
}
func (m *metadataManager) opMetaInodeAccessTimeGet(conn net.Conn, p *Packet,
remoteAddr string) (err error) {
remoteAddr string,
) (err error) {
req := &InodeGetReq{}
if err = json.Unmarshal(p.Data, req); err != nil {
p.PacketErrorWithBody(proto.OpErr, ([]byte)(err.Error()))

4
metanode/partition.go
View File

@ -898,10 +898,6 @@ func NewMetaPartition(conf *MetaPartitionConfig, manager *metadataManager) MetaP
return mp
}
func (mp *metaPartition) GetVolName() (volName string) {
return mp.config.VolName
}
func (mp *metaPartition) GetVerSeq() uint64 {
return atomic.LoadUint64(&mp.verSeq)
}

3
metanode/partition_fsmop_inode.go
View File

@ -19,7 +19,6 @@ import (
"encoding/binary"
"encoding/json"
"fmt"
"github.com/cubefs/cubefs/util/timeutil"
"io"
"time"
@ -195,7 +194,7 @@ func (mp *metaPartition) getInode(ino *Inode, listAll bool) (resp *InodeResponse
// if ctime > i.AccessTime {
// i.AccessTime = ctime
// }
respIno := item.Copy().(*Inode)
respIno := i.Copy().(*Inode)
respIno.AccessTime = timeutil.GetCurrentTimeUnix()
resp.Msg = respIno
return

4
metanode/partition_op_extent.go
View File

@ -22,6 +22,8 @@ import (
"sort"
"time"
"github.com/cubefs/cubefs/util/timeutil"
"github.com/cubefs/cubefs/proto"
"github.com/cubefs/cubefs/util/auditlog"
"github.com/cubefs/cubefs/util/errors"
@ -648,7 +650,7 @@ func (mp *metaPartition) persistInodeAccessTime(inode uint64, p *Packet) {
return
}
ino := item.(*Inode)
ctime := Now.GetCurrentTimeUnix()
ctime := timeutil.GetCurrentTimeUnix()
at := time.Unix(ino.AccessTime, 0)
if !(ctime > ino.AccessTime && time.Now().Sub(at) > mp.GetAccessTimeValidInterval()*time.Second) {
log.LogDebugf("%v %v %v", ctime > ino.AccessTime,

2
proto/admin_proto.go
View File

@ -250,7 +250,7 @@ const (
QuotaGet = "/quota/get"
// QuotaBatchModifyPath = "/quota/batchModifyPath"
QuotaListAll = "/quota/listAll"
//trash
// trash
AdminSetTrashInterval = "/vol/setTrashInterval"
AdminSetVolAccessTimeValidInterval = "/vol/setAccessTimeValidInterval"

12
proto/packet.go
View File

@ -255,10 +255,9 @@ const (
OpTxRollbackErr uint8 = 0xED
OpTxUnknownOp uint8 = 0xEE
// multiVersion to dp/mp
OpVersionOperation uint8 = 0xD5
OpSplitMarkDelete uint8 = 0xD6
OpTryOtherExtent uint8 = 0xD7
OpReadRepairExtentAgain uint8 = 0xEF
OpVersionOperation uint8 = 0xD5
OpSplitMarkDelete uint8 = 0xD6
OpTryOtherExtent uint8 = 0xD7
// io speed limit
OpLimitedIoErr uint8 = 0xB1
@ -267,9 +266,8 @@ const (
// get access time
OpMetaInodeAccessTimeGet uint8 = 0xB2
OpLimitedIoErr uint8 = 0xB1
OpStoreClosed uint8 = 0xB4
OpReachMaxExtentsErr uint8 = 0xB3
OpReachMaxExtentsErr uint8 = 0xB3
)
const (

24
sdk/data/blobstore/reader_test.go
View File

@ -471,7 +471,8 @@ func TestReadSliceRange(t *testing.T) {
}
func MockGetObjExtentsTrue(m *meta.MetaWrapper, inode uint64) (gen uint64, size uint64,
extents []proto.ExtentKey, objExtents []proto.ObjExtentKey, err error) {
extents []proto.ExtentKey, objExtents []proto.ObjExtentKey, err error,
) {
objEks := make([]proto.ObjExtentKey, 0)
objEkLen := 5
expectedFileSize := 0
@ -484,13 +485,15 @@ func MockGetObjExtentsTrue(m *meta.MetaWrapper, inode uint64) (gen uint64, size
}
func MockGetObjExtentsFalse(m *meta.MetaWrapper, inode uint64) (gen uint64, size uint64,
extents []proto.ExtentKey, objExtents []proto.ObjExtentKey, err error) {
extents []proto.ExtentKey, objExtents []proto.ObjExtentKey, err error,
) {
return 1, 1, nil, nil, errors.New("Get objEks failed")
}
func MockEbscReadTrue(ebsc *BlobStoreClient, ctx context.Context, volName string,
buf []byte, offset uint64, size uint64,
oek proto.ObjExtentKey) (readN int, err error) {
oek proto.ObjExtentKey,
) (readN int, err error) {
reader := strings.NewReader("Hello world.")
readN, err = io.ReadFull(reader, buf)
return readN, nil
@ -498,17 +501,20 @@ func MockEbscReadTrue(ebsc *BlobStoreClient, ctx context.Context, volName string
func MockEbscReadFalse(ebsc *BlobStoreClient, ctx context.Context, volName string,
buf []byte, offset uint64, size uint64,
oek proto.ObjExtentKey) (readN int, err error) {
oek proto.ObjExtentKey,
) (readN int, err error) {
return 0, syscall.EIO
}
func MockReadExtentTrue(client *stream.ExtentClient, inode uint64, ek *proto.ExtentKey,
data []byte, offset int, size int) (read int, err error, b bool) {
data []byte, offset int, size int,
) (read int, err error, b bool) {
return len("Hello world"), nil, true
}
func MockReadExtentFalse(client *stream.ExtentClient, inode uint64, ek *proto.ExtentKey,
data []byte, offset int, size int) (read int, err error) {
data []byte, offset int, size int,
) (read int, err error) {
return 0, errors.New("Read extent failed")
}
@ -521,12 +527,14 @@ func MockCheckDataPartitionExistFalse(client *stream.ExtentClient, partitionID u
}
func MockWriteTrue(client *stream.ExtentClient, inode uint64, offset int, data []byte,
flags int, checkFunc func() error) (write int, err error) {
flags int, checkFunc func() error,
) (write int, err error) {
return len(data), nil
}
func MockWriteFalse(client *stream.ExtentClient, inode uint64, offset int, data []byte,
flags int) (write int, err error) {
flags int,
) (write int, err error) {
return 0, errors.New("Write failed")
}