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(flashnode): turn on off client cache in master

Browse Source 
. #22115371

Signed-off-by: slasher <mcq.sejust@gmail.com>
This commit is contained in:
slasher 2023-12-26 11:59:12 +08:00 committed by zhumingze1108
parent fccb8d1947
commit 1877c03de8
8 changed files with 133 additions and 41 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

47
cli/cmd/flashgroup.go
View File

@ -15,6 +15,7 @@
package cmd
import (
"encoding/json"
"fmt"
"math"
"sort"
@ -39,6 +40,7 @@ func newFlashGroupCmd(client *master.MasterClient) *cobra.Command {
Short: "cluster flashgroup management",
}
cmd.AddCommand(
newCmdFlashGroupTurn(client),
newCmdFlashGroupCreate(client),
newCmdFlashGroupSet(client),
newCmdFlashGroupRemove(client),
@ -46,12 +48,34 @@ func newFlashGroupCmd(client *master.MasterClient) *cobra.Command {
newCmdFlashGroupNodeRemove(client),
newCmdFlashGroupGet(client),
newCmdFlashGroupList(client),
newCmdFlashGroupClient(client),
newCmdFlashGroupSearch(client),
newCmdFlashGroupGraph(client),
)
return cmd
}
func newCmdFlashGroupTurn(client *master.MasterClient) *cobra.Command {
return &cobra.Command{
Use: "turn [IsEnable]",
Short: "turn flash group cache",
Args: cobra.MinimumNArgs(1),
Run: func(cmd *cobra.Command, args []string) {
var err error
defer func() { errout(err) }()
enabled, err := strconv.ParseBool(args[0])
if err != nil {
return
}
result, err := client.AdminAPI().TurnFlashGroup(enabled)
if err != nil {
return
}
stdoutln(result)
},
}
}
func newCmdFlashGroupCreate(client *master.MasterClient) *cobra.Command {
var optSlots string
cmd := &cobra.Command{
@ -129,7 +153,7 @@ func newCmdFlashGroupNodeAdd(client *master.MasterClient) *cobra.Command {
optCount int
)
cmd := &cobra.Command{
Use: "addNode" + _flashgroupID,
Use: "nodeAdd" + _flashgroupID,
Short: "add flash node to given flash group",
Args: cobra.MinimumNArgs(1),
Run: func(cmd *cobra.Command, args []string) {
@ -159,7 +183,7 @@ func newCmdFlashGroupNodeRemove(client *master.MasterClient) *cobra.Command {
optCount int
)
cmd := &cobra.Command{
Use: "removeNode" + _flashgroupID,
Use: "nodeRemove" + _flashgroupID,
Short: "remove flash node to given flash group",
Args: cobra.MinimumNArgs(1),
Run: func(cmd *cobra.Command, args []string) {
@ -278,6 +302,25 @@ func newCmdFlashGroupList(client *master.MasterClient) *cobra.Command {
}
}
func newCmdFlashGroupClient(client *master.MasterClient) *cobra.Command {
return &cobra.Command{
Use: "client",
Short: "show client response",
Args: cobra.MinimumNArgs(0),
Run: func(cmd *cobra.Command, args []string) {
var err error
defer func() { errout(err) }()
fgv, err := client.AdminAPI().ClientFlashGroups()
if err != nil {
return
}
stdoutln("Client Response:")
b, _ := json.MarshalIndent(fgv, "", " ")
stdoutln(string(b))
},
}
}
func newCmdFlashGroupSearch(client *master.MasterClient) *cobra.Command {
return &cobra.Command{
Use: "search [volume] [inode] [offset]",

37
master/flash_group.go
View File

@ -139,6 +139,27 @@ func (fg *FlashGroup) GetAdminView() (view proto.FlashGroupAdminView) {
return
}
func (m *Server) turnFlashGroup(w http.ResponseWriter, r *http.Request) {
var err error
metric := exporter.NewTPCnt(apiToMetricsName(proto.AdminFlashGroupTurn))
defer func() {
doStatAndMetric(proto.AdminFlashGroupTurn, metric, err, nil)
}()
r.ParseForm()
enabled, err := extractStatus(r)
if err != nil {
sendErrReply(w, r, newErrHTTPReply(err))
return
}
topo := m.cluster.flashNodeTopo
if enabled {
topo.clientOff.Store([]byte(nil))
} else {
topo.clientOff.Store(topo.clientEmpty)
}
sendOkReply(w, r, newSuccessHTTPReply(fmt.Sprintf("turn %v", enabled)))
}
func (m *Server) createFlashGroup(w http.ResponseWriter, r *http.Request) {
var err error
metric := exporter.NewTPCnt(apiToMetricsName(proto.AdminFlashGroupCreate))
@ -465,24 +486,14 @@ func (m *Server) clientFlashGroups(w http.ResponseWriter, r *http.Request) {
defer func() {
doStatAndMetric(proto.ClientFlashGroups, metric, err, nil)
}()
cache, err := m.cluster.getFlashGroupClientResponseCache()
if err != nil {
sendErrReply(w, r, newErrHTTPReply(err))
cache := m.cluster.flashNodeTopo.getClientResponse()
if len(cache) == 0 {
sendErrReply(w, r, newErrHTTPReply(fmt.Errorf("flash group response cache is empty")))
return
}
send(w, r, cache)
}
func (c *Cluster) getFlashGroupClientResponseCache() (cache []byte, err error) {
if cache := c.flashNodeTopo.clientRespCache.Load().([]byte); len(cache) == 0 {
c.updateFlashGroupClientCache()
}
if cache = c.flashNodeTopo.clientRespCache.Load().([]byte); len(cache) == 0 {
return nil, fmt.Errorf("flash group resp cache is empty")
}
return
}
func parseRequestForManageFlashNodeOfFlashGroup(r *http.Request) (flashGroupID uint64, addr, zoneName string, count int, err error) {
if flashGroupID, err = extractFlashGroupID(r); err != nil {
return

10
master/flash_group_test.go
View File

@ -52,6 +52,7 @@ func removeFlashGroups(t *testing.T, groups []proto.FlashGroupAdminView) {
}
func testFlashGroup(t *testing.T) {
t.Run("Turn", testFlashGroupTurn)
t.Run("Create", testFlashGroupCreate)
t.Run("Set", testFlashGroupSet)
t.Run("Remove", testFlashGroupRemove)
@ -61,6 +62,15 @@ func testFlashGroup(t *testing.T) {
t.Run("Client", testFlashGroupClient)
}
func testFlashGroupTurn(t *testing.T) {
groups := createFlashGroups(t)
defer removeFlashGroups(t, groups)
_, err := mc.AdminAPI().TurnFlashGroup(false)
require.NoError(t, err)
_, err = mc.AdminAPI().TurnFlashGroup(true)
require.NoError(t, err)
}
func testFlashGroupCreate(t *testing.T) {
groups := createFlashGroups(t)
defer removeFlashGroups(t, groups)

71
master/flash_node_topology.go
View File

@ -38,8 +38,10 @@ type flashNodeTopology struct {
flashNodeMap sync.Map // key: FlashNodeAddr, value: *FlashNode
zoneMap sync.Map // key: zoneName, value: *FlashNodeZone
clientUpdateCh chan struct{}
clientRespCache atomic.Value // []byte
clientEmpty []byte // empty response cache
clientOff atomic.Value // []byte, default nil (on)
clientCache atomic.Value // []byte
clientUpdateCh chan struct{} // update client response cache
}
type FlashNodeZone struct {
@ -80,9 +82,17 @@ func newFlashNodeZone(name string) (zone *FlashNodeZone) {
}
func newFlashNodeTopology() (t *flashNodeTopology) {
t = &flashNodeTopology{slotsMap: make(map[uint32]uint64)}
t.clientUpdateCh = make(chan struct{}, 1)
t.clientRespCache.Store([]byte(nil))
empty, err := json.Marshal(newSuccessHTTPReply(proto.FlashGroupView{}))
if err != nil {
panic(fmt.Sprintf("action[newFlashNodeTopology] json marshal %v", err))
}
t = &flashNodeTopology{
slotsMap: make(map[uint32]uint64),
clientEmpty: empty,
clientUpdateCh: make(chan struct{}, 1),
}
t.clientOff.Store([]byte(nil))
t.clientCache.Store([]byte(nil))
return t
}
@ -101,14 +111,7 @@ func (t *flashNodeTopology) clear() {
flashNode.clean()
return true
})
t.clientRespCache.Store([]byte(nil))
}
func (t *flashNodeTopology) updateClientCache() {
select {
case t.clientUpdateCh <- struct{}{}:
default:
}
t.clientCache.Store([]byte(nil))
}
func (t *flashNodeTopology) getZone(name string) (zone *FlashNodeZone, err error) {
@ -239,6 +242,7 @@ func (t *flashNodeTopology) getFlashGroup(fgID uint64) (flashGroup *FlashGroup,
func (t *flashNodeTopology) getFlashGroupView() (fgv *proto.FlashGroupView) {
fgv = new(proto.FlashGroupView)
fgv.Enable = true
t.flashGroupMap.Range(func(_, value interface{}) bool {
fg := value.(*FlashGroup)
if fg.GetStatus().IsActive() {
@ -269,6 +273,33 @@ func (t *flashNodeTopology) getFlashGroupsAdminView(fgStatus proto.FlashGroupSta
return
}
func (t *flashNodeTopology) updateClientCache() {
select {
case t.clientUpdateCh <- struct{}{}:
default:
}
}
func (t *flashNodeTopology) getClientResponse() []byte {
if cache := t.clientOff.Load().([]byte); len(cache) > 0 {
return cache
}
if cache := t.clientCache.Load().([]byte); len(cache) > 0 {
return cache
}
return t.updateClientResponse()
}
func (t *flashNodeTopology) updateClientResponse() []byte {
cache, err := json.Marshal(newSuccessHTTPReply(t.getFlashGroupView()))
if err != nil {
log.LogError("action[updateClientResponse] json marshal", err)
return nil
}
t.clientCache.Store(cache)
return cache[:]
}
func (c *Cluster) loadFlashNodes() (err error) {
result, err := c.fsm.store.SeekForPrefix([]byte(flashNodePrefix))
if err != nil {
@ -345,7 +376,7 @@ func (c *Cluster) scheduleToUpdateFlashGroupRespCache() {
defer ticker.Stop()
for {
if c.partition != nil && c.partition.IsRaftLeader() {
c.updateFlashGroupClientCache()
c.flashNodeTopo.updateClientResponse()
}
select {
case <-c.flashNodeTopo.clientUpdateCh:
@ -355,15 +386,3 @@ func (c *Cluster) scheduleToUpdateFlashGroupRespCache() {
}
}()
}
func (c *Cluster) updateFlashGroupClientCache() {
fgv := c.flashNodeTopo.getFlashGroupView()
reply := newSuccessHTTPReply(fgv)
cache, err := json.Marshal(reply)
if err != nil {
msg := fmt.Sprintf("action[updateFlashGroupClientCache] json marshal err:%v", err)
log.LogError(msg)
return
}
c.flashNodeTopo.clientRespCache.Store(cache)
}

1
master/http_server.go
View File

@ -859,6 +859,7 @@ func (m *Server) registerAPIRoutes(router *mux.Router) {
router.NewRoute().Methods(http.MethodGet).Path(proto.FlashNodeList).HandlerFunc(m.listFlashNodes)
// APIs for FlashGroup
router.NewRoute().Methods(http.MethodPost).Path(proto.AdminFlashGroupTurn).HandlerFunc(m.turnFlashGroup)
router.NewRoute().Methods(http.MethodPost).Path(proto.AdminFlashGroupCreate).HandlerFunc(m.createFlashGroup)
router.NewRoute().Methods(http.MethodPost).Path(proto.AdminFlashGroupSet).HandlerFunc(m.setFlashGroup)
router.NewRoute().Methods(http.MethodPost).Path(proto.AdminFlashGroupRemove).HandlerFunc(m.removeFlashGroup)

1
proto/admin_proto.go
View File

@ -281,6 +281,7 @@ const (
FlashNodeList = "/flashNode/list"
// FlashGroup API
AdminFlashGroupTurn = "/flashGroup/turn"
AdminFlashGroupCreate = "/flashGroup/create"
AdminFlashGroupSet = "/flashGroup/set"
AdminFlashGroupRemove = "/flashGroup/remove"

1
proto/distributed_cache.go
View File

@ -44,6 +44,7 @@ type FlashGroupInfo struct {
}
type FlashGroupView struct {
Enable bool
FlashGroups []*FlashGroupInfo
}

6
sdk/master/api_admin.go
View File

@ -920,6 +920,12 @@ func (api *AdminAPI) ChangeMasterLeader(leaderAddr string) (err error) {
return
}
func (api *AdminAPI) TurnFlashGroup(enable bool) (result string, err error) {
request := newRequest(post, proto.AdminFlashGroupTurn).Header(api.h).addParamAny("enable", enable)
data, err := api.mc.serveRequest(request)
return string(data), err
}
func (api *AdminAPI) CreateFlashGroup(slots string) (fgView proto.FlashGroupAdminView, err error) {
err = api.mc.requestWith(&fgView, newRequest(post, proto.AdminFlashGroupCreate).
Header(api.h).addParam("slots", slots))