diff --git a/cli/cmd/cluster.go b/cli/cmd/cluster.go index 178989761..443d96c93 100644 --- a/cli/cmd/cluster.go +++ b/cli/cmd/cluster.go @@ -289,6 +289,12 @@ func newClusterSetParasCmd(client *master.MasterClient) *cobra.Command { dataMediaType := "" handleTimeout := "" readDataNodeTimeout := "" + optRcTTL := "" + optRcReadTimeout := "" + optRemoteCacheMultiRead := "" + optFlashNodeTimeoutCount := "" + optRemoteCacheSameZoneTimeout := "" + optRemoteCacheSameRegionTimeout := "" cmd := &cobra.Command{ Use: CliOpSetCluster, Short: cmdClusterSetClusterInfoShort, @@ -423,12 +429,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") @@ -462,6 +527,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 } diff --git a/proto/admin_proto.go b/proto/admin_proto.go index 29edf1319..6068f164d 100644 --- a/proto/admin_proto.go +++ b/proto/admin_proto.go @@ -104,6 +104,7 @@ const ( AdminAbortDecommissionDisk = "/admin/abortDecommissionDisk" AdminResetDataPartitionRestoreStatus = "/admin/resetDataPartitionRestoreStatus" AdminGetOpLog = "/admin/getOpLog" + AdminGetRemoteCacheConfig = "/admin/getRemoteCacheConfig" // #nosec G101 AdminQueryDecommissionToken = "/admin/queryDecommissionToken" diff --git a/proto/distributed_cache.go b/proto/distributed_cache.go index 6f5693b67..7496b894b 100644 --- a/proto/distributed_cache.go +++ b/proto/distributed_cache.go @@ -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 diff --git a/remotecache/flashgroupmanager/api_args_parse.go b/remotecache/flashgroupmanager/api_args_parse.go index 804adf559..125e1ced3 100644 --- a/remotecache/flashgroupmanager/api_args_parse.go +++ b/remotecache/flashgroupmanager/api_args_parse.go @@ -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 diff --git a/remotecache/flashgroupmanager/api_service.go b/remotecache/flashgroupmanager/api_service.go index 95df2bc72..d59d3505e 100644 --- a/remotecache/flashgroupmanager/api_service.go +++ b/remotecache/flashgroupmanager/api_service.go @@ -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 diff --git a/remotecache/flashgroupmanager/cluster.go b/remotecache/flashgroupmanager/cluster.go index c4edee646..0c15ca588 100644 --- a/remotecache/flashgroupmanager/cluster.go +++ b/remotecache/flashgroupmanager/cluster.go @@ -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 }) diff --git a/remotecache/flashgroupmanager/config.go b/remotecache/flashgroupmanager/config.go index 3e5c9a2ba..c609920c5 100644 --- a/remotecache/flashgroupmanager/config.go +++ b/remotecache/flashgroupmanager/config.go @@ -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 } diff --git a/remotecache/flashgroupmanager/http_server.go b/remotecache/flashgroupmanager/http_server.go index da2a29539..ade401e4e 100644 --- a/remotecache/flashgroupmanager/http_server.go +++ b/remotecache/flashgroupmanager/http_server.go @@ -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()) diff --git a/remotecache/flashgroupmanager/metadata_fsm_op.go b/remotecache/flashgroupmanager/metadata_fsm_op.go index 5058ae336..be6a5a216 100644 --- a/remotecache/flashgroupmanager/metadata_fsm_op.go +++ b/remotecache/flashgroupmanager/metadata_fsm_op.go @@ -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 diff --git a/sdk/master/api_admin.go b/sdk/master/api_admin.go index 7f77fa0e1..2d1fa20d7 100644 --- a/sdk/master/api_admin.go +++ b/sdk/master/api_admin.go @@ -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 } @@ -1047,3 +1068,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 +} diff --git a/sdk/remotecache/client.go b/sdk/remotecache/client.go index 07c8bdb8d..ec5880539 100755 --- a/sdk/remotecache/client.go +++ b/sdk/remotecache/client.go @@ -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 }