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(flashgroupmanager): enable set remotecache config.#1000181329

Browse Source 
Signed-off-by: baihailong <baihailong@oppo.com>
This commit is contained in:
baihailong 2025-06-17 16:58:09 +08:00 committed by chihe
parent bf87156ac3
commit 84192cb594
11 changed files with 458 additions and 35 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

74
cli/cmd/cluster.go
View File

@ -288,6 +288,12 @@ func newClusterSetParasCmd(client *master.MasterClient) *cobra.Command {
dataMediaType := ""
handleTimeout := ""
readDataNodeTimeout := ""
optRcTTL := ""
optRcReadTimeout := ""
optRemoteCacheMultiRead := ""
optFlashNodeTimeoutCount := ""
optRemoteCacheSameZoneTimeout := ""
optRemoteCacheSameRegionTimeout := ""
cmd := &cobra.Command{
Use: CliOpSetCluster,
Short: cmdClusterSetClusterInfoShort,
@ -422,12 +428,71 @@ func newClusterSetParasCmd(client *master.MasterClient) *cobra.Command {
}
}
if optRcTTL != "" {
if tmp, err = strconv.ParseInt(optRcTTL, 10, 64); err != nil {
err = fmt.Errorf("param (%v) failed, should be int", optRcTTL)
return
}
if tmp <= cmdVolMinRemoteCacheTTL {
err = fmt.Errorf("param remoteCacheTTL(%v) must greater than or equal to %v", optRcTTL, cmdVolMinRemoteCacheTTL)
return
}
}
if optRcReadTimeout != "" {
if tmp, err = strconv.ParseInt(optRcReadTimeout, 10, 64); err != nil {
err = fmt.Errorf("param (%v) failed, should be int", optRcReadTimeout)
return
}
if tmp <= 0 {
err = fmt.Errorf("param remoteCacheReadTimeout(%v) must greater than 0", optRcReadTimeout)
return
}
}
if optRemoteCacheMultiRead != "" {
if _, err = strconv.ParseBool(optRemoteCacheMultiRead); err != nil {
err = fmt.Errorf("param remoteCacheMultiRead(%v) should be true or false", optRemoteCacheMultiRead)
return
}
}
if optFlashNodeTimeoutCount != "" {
if tmp, err = strconv.ParseInt(optFlashNodeTimeoutCount, 10, 64); err != nil {
err = fmt.Errorf("param (%v) failed, should be int", optFlashNodeTimeoutCount)
return
}
if tmp <= 0 {
err = fmt.Errorf("param flashNodeTimeoutCount(%v) must greater than 0", optFlashNodeTimeoutCount)
return
}
}
if optRemoteCacheSameZoneTimeout != "" {
if tmp, err = strconv.ParseInt(optRemoteCacheSameZoneTimeout, 10, 64); err != nil {
err = fmt.Errorf("param (%v) failed, should be int", optRemoteCacheSameZoneTimeout)
return
}
if tmp <= 0 {
err = fmt.Errorf("param remoteCacheSameZoneTimeout(%v) must greater than 0", optRemoteCacheSameZoneTimeout)
return
}
}
if optRemoteCacheSameRegionTimeout != "" {
if tmp, err = strconv.ParseInt(optRemoteCacheSameRegionTimeout, 10, 64); err != nil {
err = fmt.Errorf("param (%v) failed, should be int", optRemoteCacheSameRegionTimeout)
return
}
if tmp <= 0 {
err = fmt.Errorf("param remoteCacheSameRegionTimeout(%v) must greater than 0", optRemoteCacheSameRegionTimeout)
return
}
}
if err = client.AdminAPI().SetClusterParas(optDelBatchCount, optMarkDeleteRate, optDelWorkerSleepMs,
optAutoRepairRate, optLoadFactor, opMaxDpCntLimit, opMaxMpCntLimit, clientIDKey,
autoDecommissionDisk, autoDecommissionDiskInterval,
autoDpMetaRepair, autoDpMetaRepairParallelCnt,
dpRepairTimeout, dpTimeout, mpTimeout, dpBackupTimeout, decommissionDpLimit, decommissionDiskLimit,
forbidWriteOpOfProtoVersion0, dataMediaType, handleTimeout, readDataNodeTimeout); err != nil {
forbidWriteOpOfProtoVersion0, dataMediaType, handleTimeout, readDataNodeTimeout,
optRcTTL, optRcReadTimeout, optRemoteCacheMultiRead, optFlashNodeTimeoutCount,
optRemoteCacheSameZoneTimeout, optRemoteCacheSameRegionTimeout); err != nil {
return
}
stdout("Cluster parameters has been set successfully. \n")
@ -461,6 +526,13 @@ func newClusterSetParasCmd(client *master.MasterClient) *cobra.Command {
cmd.Flags().StringVar(&dataMediaType, "clusterDataMediaType", "", "set cluster media type, 1(ssd), 2(hdd)")
cmd.Flags().StringVar(&handleTimeout, "flashNodeHandleReadTimeout", "", "Specify flash node handle read timeout (example:1000ms)")
cmd.Flags().StringVar(&readDataNodeTimeout, "flashNodeReadDataNodeTimeout", "", "Specify flash node read data node timeout (example:3000ms)")
cmd.Flags().StringVar(&optRcTTL, CliFlagRemoteCacheTTL, "", "Remote cache ttl[Unit: s](must >= 10min, default 5day)")
cmd.Flags().StringVar(&optRcReadTimeout, CliFlagRemoteCacheReadTimeout, "", "Remote cache read timeout millisecond(must > 0)")
cmd.Flags().StringVar(&optRemoteCacheMultiRead, CliFlagRemoteCacheMultiRead, "", "Remote cache follower read(true|false)")
cmd.Flags().StringVar(&optFlashNodeTimeoutCount, CliFlagFlashNodeTimeoutCount, "", "FlashNode timeout count, flashNode will be removed by client if it's timeout count exceeds this value")
cmd.Flags().StringVar(&optRemoteCacheSameZoneTimeout, CliFlagRemoteCacheSameZoneTimeout, "", "Remote cache same zone timeout microsecond(must > 0)")
cmd.Flags().StringVar(&optRemoteCacheSameRegionTimeout, CliFlagRemoteCacheSameRegionTimeout, "", "Remote cache same region timeout millisecond(must > 0)")
return cmd
}

1
proto/admin_proto.go
View File

@ -103,6 +103,7 @@ const (
AdminAbortDecommissionDisk = "/admin/abortDecommissionDisk"
AdminResetDataPartitionRestoreStatus = "/admin/resetDataPartitionRestoreStatus"
AdminGetOpLog = "/admin/getOpLog"
AdminGetRemoteCacheConfig = "/admin/getRemoteCacheConfig"
// #nosec G101
AdminQueryDecommissionToken = "/admin/queryDecommissionToken"

11
proto/distributed_cache.go
View File

@ -374,6 +374,17 @@ type FlashNodeSlotStat struct {
SlotStat []*SlotStat
}
type RemoteCacheConfig struct {
FlashNodeHandleReadTimeout int
FlashNodeReadDataNodeTimeout int
RemoteCacheTTL int64
RemoteCacheReadTimeout int64
RemoteCacheMultiRead bool
FlashNodeTimeoutCount int64
RemoteCacheSameZoneTimeout int64
RemoteCacheSameRegionTimeout int64
}
func ComputeSourcesVersion(sources []*DataSource, gen uint64) (version uint32) {
if len(sources) == 0 {
return 0

66
remotecache/flashgroupmanager/api_args_parse.go
View File

@ -145,6 +145,72 @@ func parseAndExtractSetNodeInfoParams(r *http.Request) (params map[string]interf
params[cfgFlashNodeReadDataNodeTimeout] = val
}
if value = r.FormValue(cfgRemoteCacheTTL); value != "" {
noParams = false
val := int64(0)
val, err = strconv.ParseInt(value, 10, 32)
if err != nil {
err = unmatchedKey(cfgRemoteCacheTTL)
return
}
params[cfgRemoteCacheTTL] = val
}
if value = r.FormValue(cfgRemoteCacheReadTimeout); value != "" {
noParams = false
val := int64(0)
val, err = strconv.ParseInt(value, 10, 32)
if err != nil {
err = unmatchedKey(cfgRemoteCacheReadTimeout)
return
}
params[cfgRemoteCacheReadTimeout] = val
}
if value = r.FormValue(cfgRemoteCacheMultiRead); value != "" {
noParams = false
val := false
val, err = strconv.ParseBool(value)
if err != nil {
err = unmatchedKey(cfgRemoteCacheMultiRead)
return
}
params[cfgRemoteCacheMultiRead] = val
}
if value = r.FormValue(cfgFlashNodeTimeoutCount); value != "" {
noParams = false
val := int64(0)
val, err = strconv.ParseInt(value, 10, 32)
if err != nil {
err = unmatchedKey(cfgFlashNodeTimeoutCount)
return
}
params[cfgFlashNodeTimeoutCount] = val
}
if value = r.FormValue(cfgRemoteCacheSameZoneTimeout); value != "" {
noParams = false
val := int64(0)
val, err = strconv.ParseInt(value, 10, 32)
if err != nil {
err = unmatchedKey(cfgRemoteCacheSameZoneTimeout)
return
}
params[cfgRemoteCacheSameZoneTimeout] = val
}
if value = r.FormValue(cfgRemoteCacheSameRegionTimeout); value != "" {
noParams = false
val := int64(0)
val, err = strconv.ParseInt(value, 10, 32)
if err != nil {
err = unmatchedKey(cfgRemoteCacheSameRegionTimeout)
return
}
params[cfgRemoteCacheSameRegionTimeout] = val
}
if noParams {
err = fmt.Errorf("no key assigned")
return

160
remotecache/flashgroupmanager/api_service.go
View File

@ -57,8 +57,8 @@ func (m *FlashGroupManager) getCluster(w http.ResponseWriter, r *http.Request) {
LeaderAddr: m.leaderInfo.addr,
MasterNodes: make([]proto.NodeView, 0),
FlashNodes: make([]proto.NodeView, 0),
FlashNodeHandleReadTimeout: m.cluster.cfg.flashNodeHandleReadTimeout,
FlashNodeReadDataNodeTimeout: m.cluster.cfg.flashNodeReadDataNodeTimeout,
FlashNodeHandleReadTimeout: m.cluster.cfg.FlashNodeHandleReadTimeout,
FlashNodeReadDataNodeTimeout: m.cluster.cfg.FlashNodeReadDataNodeTimeout,
}
cv.DataNodeStatInfo = new(proto.NodeStatInfo)
cv.MetaNodeStatInfo = new(proto.NodeStatInfo)
@ -90,6 +90,25 @@ func (m *FlashGroupManager) getIPAddr(w http.ResponseWriter, r *http.Request) {
sendOkReply(w, r, newSuccessHTTPReply(cInfo))
}
func (m *FlashGroupManager) getRemoteCacheConfig(w http.ResponseWriter, r *http.Request) {
var err error
metric := exporter.NewTPCnt(apiToMetricsName(proto.AdminGetRemoteCacheConfig))
defer func() {
doStatAndMetric(proto.AdminGetRemoteCacheConfig, metric, err, nil)
}()
config := &proto.RemoteCacheConfig{
RemoteCacheTTL: m.config.RemoteCacheTTL,
RemoteCacheReadTimeout: m.config.RemoteCacheReadTimeout,
RemoteCacheMultiRead: m.config.RemoteCacheMultiRead,
FlashNodeTimeoutCount: m.config.FlashNodeTimeoutCount,
RemoteCacheSameZoneTimeout: m.config.RemoteCacheSameZoneTimeout,
RemoteCacheSameRegionTimeout: m.config.RemoteCacheSameRegionTimeout,
}
log.LogInfof("getRemoteCacheConfig config(%v)", config)
sendOkReply(w, r, newSuccessHTTPReply(config))
}
func (m *FlashGroupManager) clientFlashGroups(w http.ResponseWriter, r *http.Request) {
var err error
metric := exporter.NewTPCnt(apiToMetricsName(proto.ClientFlashGroups))
@ -688,32 +707,139 @@ func (m *FlashGroupManager) setNodeInfoHandler(w http.ResponseWriter, r *http.Re
}
}
}
if val, ok := params[cfgRemoteCacheTTL]; ok {
if v, ok := val.(int64); ok {
if err = m.setConfig(cfgRemoteCacheTTL, strconv.FormatInt(v, 10)); err != nil {
sendErrReply(w, r, newErrHTTPReply(err))
return
}
}
}
if val, ok := params[cfgRemoteCacheReadTimeout]; ok {
if v, ok := val.(int64); ok {
if err = m.setConfig(cfgRemoteCacheReadTimeout, strconv.FormatInt(v, 10)); err != nil {
sendErrReply(w, r, newErrHTTPReply(err))
return
}
}
}
if val, ok := params[cfgRemoteCacheMultiRead]; ok {
if v, ok := val.(bool); ok {
if err = m.setConfig(cfgRemoteCacheMultiRead, strconv.FormatBool(v)); err != nil {
sendErrReply(w, r, newErrHTTPReply(err))
return
}
}
}
if val, ok := params[cfgFlashNodeTimeoutCount]; ok {
if v, ok := val.(int64); ok {
if err = m.setConfig(cfgFlashNodeTimeoutCount, strconv.FormatInt(v, 10)); err != nil {
sendErrReply(w, r, newErrHTTPReply(err))
return
}
}
}
if val, ok := params[cfgRemoteCacheSameZoneTimeout]; ok {
if v, ok := val.(int64); ok {
if err = m.setConfig(cfgRemoteCacheSameZoneTimeout, strconv.FormatInt(v, 10)); err != nil {
sendErrReply(w, r, newErrHTTPReply(err))
return
}
}
}
if val, ok := params[cfgRemoteCacheSameRegionTimeout]; ok {
if v, ok := val.(int64); ok {
if err = m.setConfig(cfgRemoteCacheSameRegionTimeout, strconv.FormatInt(v, 10)); err != nil {
sendErrReply(w, r, newErrHTTPReply(err))
return
}
}
}
sendOkReply(w, r, newSuccessHTTPReply(fmt.Sprintf("set nodeinfo params %v successfully", params)))
}
func (m *FlashGroupManager) setConfig(key string, value string) (err error) {
var (
fnHandleReadTimeout int
fnReadDataNodeTimeout int
oldIntValue int
fnHandleReadTimeout int
fnReadDataNodeTimeout int
remoteCacheTTL int64
remoteCacheReadTimeout int64
remoteCacheMultiRead bool
flashNodeTimeoutCount int64
remoteCacheSameZoneTimeout int64
remoteCacheSameRegionTimeout int64
oldIntValue int
oldInt64Value int64
oldBoolValue bool
)
log.LogInfof("set config key: %s, value: %s", key, value)
defer func() {
if err != nil {
log.LogErrorf("set config failed, key: %s, value: %s, err: %v", key, value, err)
}
}()
switch key {
case cfgFlashNodeHandleReadTimeout:
fnHandleReadTimeout, err = strconv.Atoi(value)
if err != nil {
return err
}
oldIntValue = m.config.flashNodeHandleReadTimeout
m.config.flashNodeHandleReadTimeout = fnHandleReadTimeout
oldIntValue = m.config.FlashNodeHandleReadTimeout
m.config.FlashNodeHandleReadTimeout = fnHandleReadTimeout
case cfgFlashNodeReadDataNodeTimeout:
fnReadDataNodeTimeout, err = strconv.Atoi(value)
if err != nil {
return err
}
oldIntValue = m.config.flashNodeReadDataNodeTimeout
m.config.flashNodeReadDataNodeTimeout = fnReadDataNodeTimeout
oldIntValue = m.config.FlashNodeReadDataNodeTimeout
m.config.FlashNodeReadDataNodeTimeout = fnReadDataNodeTimeout
case cfgRemoteCacheTTL:
remoteCacheTTL, err = strconv.ParseInt(value, 10, 64)
if err != nil {
return err
}
oldInt64Value = m.config.RemoteCacheTTL
m.config.RemoteCacheTTL = remoteCacheTTL
case cfgRemoteCacheReadTimeout:
remoteCacheReadTimeout, err = strconv.ParseInt(value, 10, 64)
if err != nil {
return err
}
oldInt64Value = m.config.RemoteCacheReadTimeout
m.config.RemoteCacheReadTimeout = remoteCacheReadTimeout
case cfgRemoteCacheMultiRead:
remoteCacheMultiRead, err = strconv.ParseBool(value)
if err != nil {
return err
}
oldBoolValue = m.config.RemoteCacheMultiRead
m.config.RemoteCacheMultiRead = remoteCacheMultiRead
case cfgFlashNodeTimeoutCount:
flashNodeTimeoutCount, err = strconv.ParseInt(value, 10, 64)
if err != nil {
return err
}
oldInt64Value = m.config.FlashNodeTimeoutCount
m.config.FlashNodeTimeoutCount = flashNodeTimeoutCount
case cfgRemoteCacheSameZoneTimeout:
remoteCacheSameZoneTimeout, err = strconv.ParseInt(value, 10, 64)
if err != nil {
return err
}
oldInt64Value = m.config.RemoteCacheSameZoneTimeout
m.config.RemoteCacheSameZoneTimeout = remoteCacheSameZoneTimeout
case cfgRemoteCacheSameRegionTimeout:
remoteCacheSameRegionTimeout, err = strconv.ParseInt(value, 10, 64)
if err != nil {
return err
}
oldInt64Value = m.config.RemoteCacheSameRegionTimeout
m.config.RemoteCacheSameRegionTimeout = remoteCacheSameRegionTimeout
default:
err = keyNotFound("config")
@ -723,9 +849,21 @@ func (m *FlashGroupManager) setConfig(key string, value string) (err error) {
if err = m.cluster.syncPutCluster(); err != nil {
switch key {
case cfgFlashNodeHandleReadTimeout:
m.config.flashNodeHandleReadTimeout = oldIntValue
m.config.FlashNodeHandleReadTimeout = oldIntValue
case cfgFlashNodeReadDataNodeTimeout:
m.config.flashNodeReadDataNodeTimeout = oldIntValue
m.config.FlashNodeReadDataNodeTimeout = oldIntValue
case cfgRemoteCacheTTL:
m.config.RemoteCacheTTL = oldInt64Value
case cfgRemoteCacheReadTimeout:
m.config.RemoteCacheReadTimeout = oldInt64Value
case cfgRemoteCacheMultiRead:
m.config.RemoteCacheMultiRead = oldBoolValue
case cfgFlashNodeTimeoutCount:
m.config.FlashNodeTimeoutCount = oldInt64Value
case cfgRemoteCacheSameZoneTimeout:
m.config.RemoteCacheSameZoneTimeout = oldInt64Value
case cfgRemoteCacheSameRegionTimeout:
m.config.RemoteCacheSameRegionTimeout = oldInt64Value
}
log.LogErrorf("setConfig syncPutCluster fail err %v", err)
return err

2
remotecache/flashgroupmanager/cluster.go
View File

@ -433,7 +433,7 @@ func (c *Cluster) checkFlashNodeHeartbeat() {
c.flashNodeTopo.flashNodeMap.Range(func(addr, flashNode interface{}) bool {
node := flashNode.(*FlashNode)
node.checkLiveliness()
task := node.createHeartbeatTask(c.masterAddr(), c.cfg.flashNodeHandleReadTimeout, c.cfg.flashNodeReadDataNodeTimeout)
task := node.createHeartbeatTask(c.masterAddr(), c.cfg.FlashNodeHandleReadTimeout, c.cfg.FlashNodeReadDataNodeTimeout)
tasks = append(tasks, task)
return true
})

30
remotecache/flashgroupmanager/config.go
View File

@ -7,6 +7,7 @@ import (
"strings"
"github.com/cubefs/cubefs/depends/tiglabs/raft/proto"
cfsProto "github.com/cubefs/cubefs/proto"
"github.com/cubefs/cubefs/raftstore"
)
@ -25,25 +26,36 @@ const (
const (
cfgFlashNodeHandleReadTimeout = "flashNodeHandleReadTimeout"
cfgFlashNodeReadDataNodeTimeout = "flashNodeReadDataNodeTimeout"
cfgRemoteCacheTTL = "remoteCacheTTL"
cfgRemoteCacheReadTimeout = "remoteCacheReadTimeout"
cfgRemoteCacheMultiRead = "remoteCacheMultiRead"
cfgFlashNodeTimeoutCount = "flashNodeTimeoutCount"
cfgRemoteCacheSameZoneTimeout = "remoteCacheSameZoneTimeout"
cfgRemoteCacheSameRegionTimeout = "remoteCacheSameRegionTimeout"
)
var AddrDatabase = make(map[uint64]string)
type clusterConfig struct {
flashNodeHandleReadTimeout int
flashNodeReadDataNodeTimeout int
httpProxyPoolSize uint64
heartbeatPort int64
replicaPort int64
peerAddrs []string
peers []raftstore.PeerAddress
cfsProto.RemoteCacheConfig
httpProxyPoolSize uint64
heartbeatPort int64
replicaPort int64
peerAddrs []string
peers []raftstore.PeerAddress
}
func newClusterConfig() (cfg *clusterConfig) {
cfg = new(clusterConfig)
cfg.flashNodeHandleReadTimeout = defaultFlashNodeHandleReadTimeout
cfg.flashNodeReadDataNodeTimeout = defaultFlashNodeReadDataNodeTimeout
cfg.FlashNodeHandleReadTimeout = defaultFlashNodeHandleReadTimeout
cfg.FlashNodeReadDataNodeTimeout = defaultFlashNodeReadDataNodeTimeout
cfg.RemoteCacheTTL = cfsProto.DefaultRemoteCacheTTL
cfg.RemoteCacheReadTimeout = cfsProto.DefaultRemoteCacheClientReadTimeout
cfg.FlashNodeTimeoutCount = cfsProto.DefaultFlashNodeTimeoutCount
cfg.RemoteCacheSameZoneTimeout = cfsProto.DefaultRemoteCacheSameZoneTimeout
cfg.RemoteCacheSameRegionTimeout = cfsProto.DefaultRemoteCacheSameRegionTimeout
return
}

11
remotecache/flashgroupmanager/http_server.go
View File

@ -70,6 +70,9 @@ func (m *FlashGroupManager) registerAPIRoutes(router *mux.Router) {
Path(proto.RaftStatus).
HandlerFunc(m.getRaftStatus)
// remoteCache config
router.NewRoute().Methods(http.MethodGet).Path(proto.AdminGetRemoteCacheConfig).HandlerFunc(m.getRemoteCacheConfig)
// APIs for FlashGroup
router.NewRoute().Methods(http.MethodGet, http.MethodPost).Path(proto.AdminFlashGroupTurn).HandlerFunc(m.turnFlashGroup)
router.NewRoute().Methods(http.MethodGet, http.MethodPost).Path(proto.AdminFlashGroupCreate).HandlerFunc(m.createFlashGroup)
@ -100,15 +103,15 @@ func (m *FlashGroupManager) registerAPIMiddleware(route *mux.Router) {
func(w http.ResponseWriter, r *http.Request) {
log.LogDebugf("action[interceptor] request, method[%v] path[%v] query[%v]", r.Method, r.URL.Path, r.URL.Query())
//TODO
//if m.partition.IsRaftLeader() {
// TODO
// if m.partition.IsRaftLeader() {
// if err := m.cluster.apiLimiter.Wait(r.URL.Path); err != nil {
// log.LogWarnf("action[interceptor] too many requests, path[%v]", r.URL.Path)
// errMsg := fmt.Sprintf("too many requests for api: %s", html.EscapeString(r.URL.Path))
// http.Error(w, errMsg, http.StatusTooManyRequests)
// return
// }
//} else {
// } else {
// if m.cluster.apiLimiter.IsFollowerLimiter(r.URL.Path) {
// if err := m.cluster.apiLimiter.Wait(r.URL.Path); err != nil {
// log.LogWarnf("action[interceptor] too many requests, path[%v]", r.URL.Path)
@ -117,7 +120,7 @@ func (m *FlashGroupManager) registerAPIMiddleware(route *mux.Router) {
// return
// }
// }
//}
// }
log.LogInfof("action[interceptor] request, remote[%v] method[%v] path[%v] query[%v]",
r.RemoteAddr, r.Method, r.URL.Path, r.URL.Query())

50
remotecache/flashgroupmanager/metadata_fsm_op.go
View File

@ -5,6 +5,7 @@ import (
"fmt"
"strings"
"github.com/cubefs/cubefs/proto"
"github.com/cubefs/cubefs/util/errors"
"github.com/cubefs/cubefs/util/log"
)
@ -28,13 +29,25 @@ type clusterValue struct {
Name string
FlashNodeHandleReadTimeout int
FlashNodeReadDataNodeTimeout int
RemoteCacheTTL int64
RemoteCacheReadTimeout int64
RemoteCacheMultiRead bool
FlashNodeTimeoutCount int64
RemoteCacheSameZoneTimeout int64
RemoteCacheSameRegionTimeout int64
}
func newClusterValue(c *Cluster) (cv *clusterValue) {
cv = &clusterValue{
Name: c.Name,
FlashNodeHandleReadTimeout: c.cfg.flashNodeHandleReadTimeout,
FlashNodeReadDataNodeTimeout: c.cfg.flashNodeReadDataNodeTimeout,
FlashNodeHandleReadTimeout: c.cfg.FlashNodeHandleReadTimeout,
FlashNodeReadDataNodeTimeout: c.cfg.FlashNodeReadDataNodeTimeout,
RemoteCacheTTL: c.cfg.RemoteCacheTTL,
RemoteCacheReadTimeout: c.cfg.RemoteCacheReadTimeout,
RemoteCacheMultiRead: c.cfg.RemoteCacheMultiRead,
FlashNodeTimeoutCount: c.cfg.FlashNodeTimeoutCount,
RemoteCacheSameZoneTimeout: c.cfg.RemoteCacheSameZoneTimeout,
RemoteCacheSameRegionTimeout: c.cfg.RemoteCacheSameRegionTimeout,
}
return cv
}
@ -63,14 +76,41 @@ func (c *Cluster) loadClusterValue() (err error) {
if cv.FlashNodeHandleReadTimeout == 0 {
cv.FlashNodeHandleReadTimeout = defaultFlashNodeHandleReadTimeout
}
c.cfg.flashNodeHandleReadTimeout = cv.FlashNodeHandleReadTimeout
c.cfg.FlashNodeHandleReadTimeout = cv.FlashNodeHandleReadTimeout
if cv.FlashNodeReadDataNodeTimeout == 0 {
cv.FlashNodeReadDataNodeTimeout = defaultFlashNodeReadDataNodeTimeout
}
c.cfg.flashNodeReadDataNodeTimeout = cv.FlashNodeReadDataNodeTimeout
c.cfg.FlashNodeReadDataNodeTimeout = cv.FlashNodeReadDataNodeTimeout
log.LogInfof("action[loadClusterValue] flashNodeHandleReadTimeout %v(ms), flashNodeReadDataNodeTimeout%v(ms)",
cv.FlashNodeHandleReadTimeout, cv.FlashNodeReadDataNodeTimeout)
if cv.RemoteCacheTTL == 0 {
cv.RemoteCacheTTL = proto.DefaultRemoteCacheTTL
}
c.cfg.RemoteCacheTTL = cv.RemoteCacheTTL
if cv.RemoteCacheReadTimeout == 0 {
cv.RemoteCacheReadTimeout = proto.DefaultRemoteCacheClientReadTimeout
}
c.cfg.RemoteCacheReadTimeout = cv.RemoteCacheReadTimeout
c.cfg.RemoteCacheMultiRead = cv.RemoteCacheMultiRead
if cv.FlashNodeTimeoutCount == 0 {
cv.FlashNodeTimeoutCount = proto.DefaultFlashNodeTimeoutCount
}
c.cfg.FlashNodeTimeoutCount = cv.FlashNodeTimeoutCount
if cv.RemoteCacheSameZoneTimeout == 0 {
cv.RemoteCacheSameZoneTimeout = proto.DefaultRemoteCacheSameZoneTimeout
}
c.cfg.RemoteCacheSameZoneTimeout = cv.RemoteCacheSameZoneTimeout
if cv.RemoteCacheSameRegionTimeout == 0 {
cv.RemoteCacheSameRegionTimeout = proto.DefaultRemoteCacheSameRegionTimeout
}
c.cfg.RemoteCacheSameRegionTimeout = cv.RemoteCacheSameRegionTimeout
log.LogInfof("action[loadClusterValue] remoteCacheTTL(%v), remoteCacheReadTimeout(%v), remoteCacheMultiRead(%v), flashNodeTimeoutCount(%v), remoteCacheSameZoneTimeout(%v), remoteCacheSameRegionTimeout(%v)",
cv.RemoteCacheTTL, cv.RemoteCacheReadTimeout, cv.RemoteCacheMultiRead, cv.FlashNodeTimeoutCount, cv.RemoteCacheSameZoneTimeout, cv.RemoteCacheSameRegionTimeout)
}
return

29
sdk/master/api_admin.go
View File

@ -599,6 +599,9 @@ func (api *AdminAPI) SetClusterParas(batchCount, markDeleteRate, deleteWorkerSle
dpRepairTimeout string, dpTimeout string, mpTimeout string, dpBackupTimeout string,
decommissionDpLimit, decommissionDiskLimit, forbidWriteOpOfProtoVersion0 string, mediaType string,
handleTimeout string, readDataNodeTimeout string,
remoteCacheTTL string, remoteCacheReadTimeout string,
remoteCacheMultiRead string, flashNodeTimeoutCount string,
remoteCacheSameZoneTimeout string, remoteCacheSameRegionTimeout string,
) (err error) {
request := newRequest(get, proto.AdminSetNodeInfo).Header(api.h)
request.addParam("batchCount", batchCount)
@ -659,7 +662,25 @@ func (api *AdminAPI) SetClusterParas(batchCount, markDeleteRate, deleteWorkerSle
if readDataNodeTimeout != "" {
request.addParam("flashNodeReadDataNodeTimeout", readDataNodeTimeout)
}
// remoteCache config
if remoteCacheTTL != "" {
request.addParamAny("remoteCacheTTL", remoteCacheTTL)
}
if remoteCacheReadTimeout != "" {
request.addParamAny("remoteCacheReadTimeout", remoteCacheReadTimeout)
}
if remoteCacheMultiRead != "" {
request.addParamAny("remoteCacheMultiRead", remoteCacheMultiRead)
}
if flashNodeTimeoutCount != "" {
request.addParamAny("flashNodeTimeoutCount", flashNodeTimeoutCount)
}
if remoteCacheSameZoneTimeout != "" {
request.addParamAny("remoteCacheSameZoneTimeout", remoteCacheSameZoneTimeout)
}
if remoteCacheSameRegionTimeout != "" {
request.addParamAny("remoteCacheSameRegionTimeout", remoteCacheSameRegionTimeout)
}
_, err = api.mc.serveRequest(request)
return
}
@ -1041,3 +1062,9 @@ func (api *AdminAPI) DeleteMetaNodeBalanceTask() (result string, err error) {
err = api.mc.requestWith(&result, newRequest(get, proto.DeleteMetaNodeBalanceTask).Header(api.h))
return
}
func (api *AdminAPI) GetRemoteCacheConfig() (config *proto.RemoteCacheConfig, err error) {
config = &proto.RemoteCacheConfig{}
err = api.mc.requestWith(config, newRequest(get, proto.AdminGetRemoteCacheConfig).Header(api.h))
return
}

59
sdk/remotecache/client.go
View File

@ -104,6 +104,10 @@ func NewRemoteCacheClient(masters []string, blockSize uint64, needInitLog bool,
}
}
if proto.Buffers == nil {
proto.InitBufferPool(32768)
}
rc = new(RemoteCacheClient)
log.LogDebugf("NewRemoteCacheClient")
rc.stopC = make(chan struct{})
@ -114,7 +118,7 @@ func NewRemoteCacheClient(masters []string, blockSize uint64, needInitLog bool,
rc.mc = master.NewMasterClient(masters, false)
rc.conns = util.NewConnectPoolWithTimeoutAndCap(5, 500, ConnIdelTimeout, 1)
rc.TTL = proto.DefaultRemoteCacheTTL
rc.RemoteCacheMultiRead = true
rc.RemoteCacheMultiRead = false
rc.FlashNodeTimeoutCount = proto.DefaultFlashNodeTimeoutCount
rc.SameZoneTimeout = proto.DefaultRemoteCacheSameZoneTimeout
rc.SameRegionTimeout = proto.DefaultRemoteCacheSameRegionTimeout
@ -122,7 +126,13 @@ func NewRemoteCacheClient(masters []string, blockSize uint64, needInitLog bool,
err = rc.UpdateFlashGroups()
if err != nil {
log.LogDebugf("NewRemoteCacheClient: updateFlashGroups err %v", err)
log.LogErrorf("NewRemoteCacheClient: updateFlashGroups err %v", err)
return
}
err = rc.updateRemoteCacheConfig()
if err != nil {
log.LogErrorf("NewRemoteCacheClient: updateRemoteCacheConfig err %v", err)
return
}
@ -178,6 +188,9 @@ func (rc *RemoteCacheClient) refreshWithRecover() (panicErr error) {
if err = rc.UpdateFlashGroups(); err != nil {
log.LogErrorf("updateFlashGroups err: %v", err)
}
if err = rc.updateRemoteCacheConfig(); err != nil {
log.LogErrorf("updateRemoteCacheConfig err: %v", err)
}
case <-refreshLatency.C:
rc.refreshHostLatency()
refreshLatency.Reset(RefreshHostLatencyInterval)
@ -775,7 +788,11 @@ func (rc *RemoteCacheClient) getReadReply(conn *net.TCPConn, reqPacket *proto.Pa
return
}
func (rc *RemoteCacheClient) Get(ctx context.Context, key string, reqId string, from, to int64) (r io.ReadCloser, length int64, shouldCache bool, err error) {
func (rc *RemoteCacheClient) Name() string {
return "remoteCache"
}
func (rc *RemoteCacheClient) Get(ctx context.Context, reqId, key string, from, to int64) (r io.ReadCloser, length int64, shouldCache bool, err error) {
log.LogDebugf("RemoteCacheClient Get: key(%v) reqId(%v) from(%v) to(%v)", key, reqId, from, to)
if from < 0 || to-from <= 0 {
return nil, 0, false, fmt.Errorf("invalid range: from(%v) to(%v)", from, to)
@ -788,7 +805,43 @@ func (rc *RemoteCacheClient) Get(ctx context.Context, key string, reqId string,
} else if strings.Contains(err.Error(), proto.ErrorNotExistShouldCache.Error()) {
shouldCache = true
}
return
}
func (rc *RemoteCacheClient) updateRemoteCacheConfig() (err error) {
var config *proto.RemoteCacheConfig
if config, err = rc.mc.AdminAPI().GetRemoteCacheConfig(); err != nil {
log.LogWarnf("updateRemoteCacheConfig: GetRemoteCacheConfig fail err(%v)", err)
return
}
log.LogInfof("updateRemoteCacheConfig: config(%v)", config)
if rc.TTL != config.RemoteCacheTTL {
log.LogInfof("updateRemoteCacheConfig RcTTL: %d -> %d", rc.TTL, config.RemoteCacheTTL)
rc.TTL = config.RemoteCacheTTL
}
if rc.ReadTimeout != config.RemoteCacheReadTimeout {
log.LogInfof("updateRemoteCacheConfig RcReadTimeoutSec: %d(ms) -> %d(ms)", rc.ReadTimeout, config.RemoteCacheReadTimeout)
rc.ReadTimeout = config.RemoteCacheReadTimeout
}
if rc.RemoteCacheMultiRead != config.RemoteCacheMultiRead {
log.LogInfof("updateRemoteCacheConfig RcMultiRead: %v -> %v", rc.RemoteCacheMultiRead, config.RemoteCacheMultiRead)
rc.RemoteCacheMultiRead = config.RemoteCacheMultiRead
}
if rc.FlashNodeTimeoutCount != int32(config.FlashNodeTimeoutCount) {
log.LogInfof("updateRemoteCacheConfig RcFlashNodeTimeoutCount: %d -> %d", rc.FlashNodeTimeoutCount, int32(config.FlashNodeTimeoutCount))
rc.FlashNodeTimeoutCount = int32(config.FlashNodeTimeoutCount)
}
if rc.SameZoneTimeout != config.RemoteCacheSameZoneTimeout {
log.LogInfof("updateRemoteCacheConfig RcSameZoneTimeout: %d -> %d", rc.SameZoneTimeout, config.RemoteCacheSameZoneTimeout)
rc.SameZoneTimeout = config.RemoteCacheSameZoneTimeout
}
if rc.SameRegionTimeout != config.RemoteCacheSameRegionTimeout {
log.LogInfof("updateRemoteCacheConfig RcSameRegionTimeout: %d -> %d", rc.SameRegionTimeout, config.RemoteCacheSameRegionTimeout)
rc.SameRegionTimeout = config.RemoteCacheSameRegionTimeout
}
return
}