feat(master): support query all client ip from master. #22834669

Signed-off-by: Victor1319 <zengxuewei@oppo.com>

cli/cli.go

@@ -19,6 +19,7 @@ import (
 	"os"
 
 	"github.com/cubefs/cubefs/cli/cmd"
+	"github.com/cubefs/cubefs/proto"
 	"github.com/cubefs/cubefs/sdk/master"
 	"github.com/cubefs/cubefs/util/log"
 	"github.com/spf13/cobra"
@@ -77,6 +78,7 @@ func main() {
 		fmt.Fprintln(os.Stderr, "Error:", err)
 		os.Exit(1)
 	}
+	proto.DumpVersion("cfs-cli")
 	if err = runCLI(); err != nil {
 		fmt.Fprintln(os.Stderr, "Error:", err)
 		log.LogError("Error:", err)

master/api_service.go

@@ -5603,6 +5603,14 @@ func (m *Server) getVol(w http.ResponseWriter, r *http.Request) {
 	}
 }
 
+// Obtain the volume information such as total capacity and used space, etc.
+func (m *Server) getAllClients(w http.ResponseWriter, r *http.Request) {
+	name, _ := extractName(r)
+	infos := m.cliMgr.GetClients(name)
+	log.LogInfof("getAllClients: get total %d client info", len(infos))
+	sendOkReply(w, r, newSuccessHTTPReply(infos))
+}
+
 // Obtain the volume information such as total capacity and used space, etc.
 func (m *Server) getVolStatInfo(w http.ResponseWriter, r *http.Request) {
 	var (
@@ -5623,11 +5631,19 @@ func (m *Server) getVolStatInfo(w http.ResponseWriter, r *http.Request) {
 		return
 	}
 
+	remoteIp := iputil.GetRealClientIP(r)
+	clientVer := r.FormValue(proto.ClientVerKey)
+	hostName := r.FormValue(proto.HostKey)
+	role := r.FormValue(proto.RoleKey)
+	log.LogInfof("getVolStatInfo: get client ver info %s, ip %s, host %s, vol %s, role %s", clientVer, remoteIp, hostName, name, role)
+
 	if vol, err = m.cluster.getVol(name); err != nil {
 		sendErrReply(w, r, newErrHTTPReply(proto.ErrVolNotExists))
 		return
 	}
 
+	m.cliMgr.PutItem(remoteIp, hostName, name, clientVer, role)
+
 	if proto.IsCold(vol.VolType) && ver != proto.LFClient {
 		sendErrReply(w, r, &proto.HTTPReply{Code: proto.ErrCodeParamError, Msg: "ec-vol is supported by LF client only"})
 		return

master/client_info.go

@@ -0,0 +1,97 @@
+package master
+
+import (
+	"fmt"
+	"strings"
+	"sync"
+	"time"
+
+	"github.com/cubefs/cubefs/util/log"
+	"github.com/cubefs/cubefs/util/timeutil"
+)
+
+const (
+	defaultVer           = "0.0.0"
+	defaultHost          = "defHost"
+	defaultRole          = "defRole"
+	clientExpireInterval = 3600
+	maxClientCnt         = 1000000
+)
+
+type ClientMgr struct {
+	sync.RWMutex
+	clients map[string]int64
+}
+
+func newClientMgr() *ClientMgr {
+	mgr := &ClientMgr{}
+	mgr.clients = make(map[string]int64)
+	go mgr.evict()
+	return mgr
+}
+
+func (cm *ClientMgr) PutItem(ip, host, vol, version, role string) {
+	cm.Lock()
+	defer cm.Unlock()
+
+	if version == "" {
+		version = defaultVer
+	}
+
+	if host == "" {
+		host = defaultHost
+	}
+
+	if role == "" {
+		role = defaultRole
+	}
+
+	key := fmt.Sprintf("_%s_%s_%s_%s_%s", vol, version, role, ip, host)
+
+	if len(cm.clients) > maxClientCnt {
+		log.LogWarnf("PutItem: too many record in cluster, ignore, key %s", key)
+		return
+	}
+
+	cm.clients[key] = timeutil.GetCurrentTimeUnix()
+}
+
+func (cm *ClientMgr) GetClients(name string) map[string]int64 {
+	cm.RLock()
+	defer cm.RUnlock()
+
+	if name != "" {
+		name = fmt.Sprintf("_%s_", name)
+	}
+
+	items := make(map[string]int64, len(cm.clients))
+	for k, v := range cm.clients {
+		if name != "" && !strings.Contains(k, name) {
+			continue
+		}
+		items[k] = v
+	}
+	return items
+}
+
+func (cm *ClientMgr) deleteByKey(key string) {
+	cm.Lock()
+	defer cm.Unlock()
+
+	delete(cm.clients, key)
+}
+
+func (cm *ClientMgr) evict() {
+	ticker := time.NewTicker(time.Minute * 10)
+	for range ticker.C {
+		log.LogInfof("ClientMgr.Evict: start check client info expire info")
+		m := cm.GetClients("")
+		now := timeutil.GetCurrentTimeUnix()
+		for k, v := range m {
+			if now > v+clientExpireInterval {
+				log.LogInfof("ClientMgr.Evict: client is expired. key %s, val %d", k, v)
+				cm.deleteByKey(k)
+			}
+		}
+	}
+}

master/http_server.go

@@ -410,6 +410,9 @@ func (m *Server) registerAPIRoutes(router *mux.Router) {
 	router.NewRoute().Methods(http.MethodGet).
 		Path(proto.ClientVolStat).
 		HandlerFunc(m.getVolStatInfo)
+	router.NewRoute().Methods(http.MethodGet).
+		Path(proto.GetAllClients).
+		HandlerFunc(m.getAllClients)
 	router.NewRoute().Methods(http.MethodGet).
 		Path(proto.GetTopologyView).
 		HandlerFunc(m.getTopology)

master/server.go

@@ -125,6 +125,7 @@ type Server struct {
 	reverseProxy    *httputil.ReverseProxy
 	metaReady       bool
 	apiServer       *http.Server
+	cliMgr          *ClientMgr
 }
 
 // NewServer creates a new server
@@ -143,6 +144,7 @@ func (m *Server) Start(cfg *config.Config) (err error) {
 		return
 	}
 	m.reverseProxy = m.newReverseProxy()
+	m.cliMgr = newClientMgr()
 
 	if m.rocksDBStore, err = raftstore_db.NewRocksDBStoreAndRecovery(m.storeDir, LRUCacheSize, WriteBufferSize); err != nil {
 		return

proto/admin_proto.go

@@ -138,6 +138,7 @@ const (
 	ClientMetaPartition      = "/metaPartition/get"
 	ClientVolStat            = "/client/volStat"
 	ClientMetaPartitions     = "/client/metaPartitions"
+	GetAllClients            = "/getAllClients"
 
 	// qos api
 	QosGetStatus           = "/qos/getStatus"
@@ -386,6 +387,9 @@ const (
 	MetaFollowerReadKey   = "metaFollowerRead"
 	LeaderRetryTimeoutKey = "leaderRetryTimeout"
 	VolEnableDirectRead   = "directRead"
+	HostKey               = "host"
+	ClientVerKey          = "clientVer"
+	RoleKey               = "role"
 )
 
 // const TimeFormat = "2006-01-02 15:04:05"

proto/version.go

@@ -10,9 +10,11 @@ var (
 	CommitID   string
 	BranchName string
 	BuildTime  string
+	Role       string
 )
 
 func DumpVersion(role string) string {
+	Role = role
 	return fmt.Sprintf("CubeFS %s\n"+
 		"Version : %s\n"+
 		"Branch  : %s\n"+

sdk/master/api_client.go

@@ -20,6 +20,7 @@ import (
 	"math/rand"
 
 	"github.com/cubefs/cubefs/proto"
+	"github.com/cubefs/cubefs/util/iputil"
 )
 
 type Decoder func([]byte) ([]byte, error)
@@ -83,7 +84,13 @@ func (api *ClientAPI) GetVolumeWithAuthnode(volName string, authKey string, toke
 func (api *ClientAPI) GetVolumeStat(volName string) (info *proto.VolStatInfo, err error) {
 	info = &proto.VolStatInfo{}
 	err = api.mc.requestWith(info, newRequest(get, proto.ClientVolStat).
-		Header(api.h).Param(anyParam{"name", volName}, anyParam{"version", proto.LFClient}))
+		Header(api.h).Param(
+		anyParam{"name", volName},
+		anyParam{"version", proto.LFClient},
+		anyParam{proto.ClientVerKey, proto.Version},
+		anyParam{proto.HostKey, iputil.HostName},
+		anyParam{proto.RoleKey, proto.Role},
+	))
 	return
 }
 

util/iputil/ip.go

@@ -18,10 +18,14 @@ import (
 	"errors"
 	"net"
 	"net/http"
+	"os"
 	"strings"
 )
 
-var cidrs []*net.IPNet
+var (
+	cidrs    []*net.IPNet
+	HostName string
+)
 
 func init() {
 	maxCidrBlocks := []string{
@@ -40,6 +44,8 @@ func init() {
 		_, cidr, _ := net.ParseCIDR(maxCidrBlock)
 		cidrs[i] = cidr
 	}
+
+	HostName, _ = os.Hostname()
 }
 
 // isLocalAddress works by checking if the address is under private CIDR blocks.
@@ -68,7 +74,6 @@ func FromRequest(r *http.Request) string {
 	// Fetch header value
 	xRealIP := r.Header.Get("X-Real-Ip")
 	xForwardedFor := r.Header.Get("X-Forwarded-For")
-
 	// If both empty, return IP from remote address
 	if xRealIP == "" && xForwardedFor == "" {
 		var remoteIP string
@@ -102,6 +107,29 @@ func RealIP(r *http.Request) string {
 	return FromRequest(r)
 }
 
+func GetRealClientIP(r *http.Request) string {
+	if xff := r.Header.Get("X-Forwarded-For"); xff != "" {
+		for _, ip := range strings.Split(xff, ",") {
+			ip = strings.TrimSpace(ip)
+			if ip != "" {
+				return ip
+			}
+		}
+	}
+
+	if xri := r.Header.Get("X-Real-IP"); xri != "" {
+		return xri
+	}
+
+	var remoteIP string
+	if strings.ContainsRune(r.RemoteAddr, ':') {
+		remoteIP, _, _ = net.SplitHostPort(r.RemoteAddr)
+	} else {
+		remoteIP = r.RemoteAddr
+	}
+	return remoteIP
+}
+
 // set default max distance from two ips to length of ipv6
 const DEFAULT_MAX_DISTANCE = 128