feat(flashgroupmanager): support api service

close:#1000151163

Signed-off-by: chihe <chihe@oppo.com>

cmd/cmd.go

@@ -40,6 +40,7 @@ import (
 	"github.com/cubefs/cubefs/metanode"
 	"github.com/cubefs/cubefs/objectnode"
 	"github.com/cubefs/cubefs/proto"
+	"github.com/cubefs/cubefs/remotecache/flashgroupmaster"
 	"github.com/cubefs/cubefs/remotecache/flashnode"
 	"github.com/cubefs/cubefs/util/auditlog"
 	"github.com/cubefs/cubefs/util/config"
@@ -67,25 +68,27 @@ const (
 )
 
 const (
-	RoleMaster    = "master"
-	RoleMeta      = "metanode"
-	RoleData      = "datanode"
-	RoleAuth      = "authnode"
-	RoleObject    = "objectnode"
-	RoleConsole   = "console"
-	RoleLifeCycle = "lcnode"
-	RoleFlash     = "flashnode"
+	RoleMaster           = "master"
+	RoleMeta             = "metanode"
+	RoleData             = "datanode"
+	RoleAuth             = "authnode"
+	RoleObject           = "objectnode"
+	RoleConsole          = "console"
+	RoleLifeCycle        = "lcnode"
+	RoleFlash            = "flashnode"
+	RoleFlashGroupMaster = "flashgroupmaster"
 )
 
 const (
-	ModuleMaster    = "master"
-	ModuleMeta      = "metaNode"
-	ModuleData      = "dataNode"
-	ModuleAuth      = "authNode"
-	ModuleObject    = "objectNode"
-	ModuleConsole   = "console"
-	ModuleLifeCycle = "lcnode"
-	ModuleFlash     = "flashNode"
+	ModuleMaster           = "master"
+	ModuleMeta             = "metaNode"
+	ModuleData             = "dataNode"
+	ModuleAuth             = "authNode"
+	ModuleObject           = "objectNode"
+	ModuleConsole          = "console"
+	ModuleLifeCycle        = "lcnode"
+	ModuleFlash            = "flashNode"
+	ModuleFlashGroupMaster = "flashNodeGroupMaster"
 )
 
 const (
@@ -224,6 +227,9 @@ func main() {
 	case RoleFlash:
 		server = flashnode.NewFlashNode()
 		module = ModuleFlash
+	case RoleFlashGroupMaster:
+		server = flashgroupmaster.NewFlashGroupMaster()
+		module = ModuleFlashGroupMaster
 	default:
 		err = errors.NewErrorf("Fatal: role mismatch: %s", role)
 		fmt.Println(err)

docker/docker-compose.yml

@@ -1,3 +1,4 @@
+version: '2.1'
 
 x-define-image: &image
     image: ${IMAGE}
@@ -56,6 +57,15 @@ x-define-flashnode: &flashnode
         - "17520"
         - 9500
 
+x-define-flashgroupmaster: &flashgroupmaster
+    <<: *servernode
+    ports:
+        - "5901"
+        - "5902"
+        - "17010"
+        - "17020"
+        - 9500
+
 x-define-command: &command
     <<: [*image, *env]
     volumes:
@@ -102,6 +112,7 @@ services:
             - flashnode1
             - flashnode2
             - flashnode3
+            - flashgroupmaster1
             - console1
             - nginx
         networks:
@@ -323,6 +334,17 @@ services:
         networks:
             extnetwork:
                 ipv4_address: 192.168.0.73
+    flashgroupmaster1:
+        <<: *flashgroupmaster
+        volumes:
+            - ${DiskPath:-./docker_data}/flashgroupmaster1/data:/cfs/data
+            - ./bin:/cfs/bin:ro
+            - ${DiskPath:-./docker_data}/flashgroupmaster1/log:/cfs/log
+            - ./conf/flashgroupmaster1.json:/cfs/conf/flashgroupmaster.json
+            - ./script/start_flashgroupmaster.sh:/cfs/script/start.sh
+        networks:
+            extnetwork:
+                ipv4_address: 192.168.0.81
 
     console1:
         <<: *servernode

remotecache/flashgroupmaster/api_args_parse.go

@@ -0,0 +1,72 @@
+package flashgroupmaster
+
+import (
+	"encoding/json"
+	"github.com/cubefs/cubefs/util/compressor"
+	"net/http"
+	"strconv"
+
+	"github.com/cubefs/cubefs/proto"
+	"github.com/cubefs/cubefs/util/log"
+)
+
+func sendErrReply(w http.ResponseWriter, r *http.Request, httpReply *proto.HTTPReply) {
+	log.LogInfof("URL[%v],remoteAddr[%v],response", r.URL, r.RemoteAddr)
+	reply, err := json.Marshal(httpReply)
+	if err != nil {
+		log.LogErrorf("fail to marshal http reply. URL[%v],remoteAddr[%v] err:[%v]", r.URL, r.RemoteAddr, err)
+		http.Error(w, "fail to marshal http reply", http.StatusBadRequest)
+		return
+	}
+
+	w.Header().Set("content-type", "application/json")
+	w.Header().Set("Content-Length", strconv.Itoa(len(reply)))
+	if _, err = w.Write(reply); err != nil {
+		log.LogErrorf("fail to write http len[%d].URL[%v],remoteAddr[%v] err:[%v]", len(reply), r.URL, r.RemoteAddr, err)
+	}
+}
+
+func newSuccessHTTPReply(data interface{}) *proto.HTTPReply {
+	return &proto.HTTPReply{Code: proto.ErrCodeSuccess, Msg: proto.ErrSuc.Error(), Data: data}
+}
+
+func newErrHTTPReply(err error) *proto.HTTPReply {
+	if err == nil {
+		return newSuccessHTTPReply("")
+	}
+
+	code, ok := proto.Err2CodeMap[err]
+	if ok {
+		return &proto.HTTPReply{Code: code, Msg: err.Error()}
+	}
+
+	return &proto.HTTPReply{Code: proto.ErrCodeInternalError, Msg: err.Error()}
+}
+
+func send(w http.ResponseWriter, r *http.Request, reply []byte) {
+	w.Header().Set("content-type", "application/json")
+	w.Header().Set("Content-Length", strconv.Itoa(len(reply)))
+	if _, err := w.Write(reply); err != nil {
+		log.LogErrorf("fail to write http len[%d].URL[%v],remoteAddr[%v] err:[%v]", len(reply), r.URL, r.RemoteAddr, err)
+		return
+	}
+}
+
+func sendOkReply(w http.ResponseWriter, r *http.Request, httpReply *proto.HTTPReply) (err error) {
+	reply, err := json.Marshal(httpReply)
+	if err != nil {
+		log.LogErrorf("fail to marshal http reply. URL[%v],remoteAddr[%v] err:[%v]", r.URL, r.RemoteAddr, err)
+		http.Error(w, "fail to marshal http reply", http.StatusBadRequest)
+		return
+	}
+
+	if acceptEncoding := r.Header.Get(proto.HeaderAcceptEncoding); acceptEncoding != "" {
+		if compressed, errx := compressor.New(acceptEncoding).Compress(reply); errx == nil {
+			w.Header().Set(proto.HeaderContentEncoding, acceptEncoding)
+			reply = compressed
+		}
+	}
+
+	send(w, r, reply)
+	return
+}

remotecache/flashgroupmaster/api_service.go

@@ -0,0 +1,79 @@
+package flashgroupmaster
+
+import (
+	"github.com/cubefs/cubefs/proto"
+	"github.com/cubefs/cubefs/util/exporter"
+	"github.com/cubefs/cubefs/util/iputil"
+	"github.com/cubefs/cubefs/util/stat"
+	"net/http"
+	"strings"
+)
+
+func apiToMetricsName(api string) (reqMetricName string) {
+	var builder strings.Builder
+	builder.WriteString("req")
+	// prometheus metric not allow '/' in name, need to transfer to '_'
+	builder.WriteString(strings.Replace(api, "/", "_", -1))
+	return builder.String()
+}
+
+func doStatAndMetric(statName string, metric *exporter.TimePointCount, err error, metricLabels map[string]string) {
+	if metric == nil {
+		return
+	}
+	if metricLabels == nil {
+		metric.Set(err)
+	} else {
+		metric.SetWithLabels(err, metricLabels)
+	}
+
+	startTime := metric.GetStartTime()
+	stat.EndStat(statName, err, &startTime, 1)
+}
+
+func (m *FlashGroupMaster) getCluster(w http.ResponseWriter, r *http.Request) {
+	metric := exporter.NewTPCnt(apiToMetricsName(proto.AdminGetCluster))
+	defer func() {
+		doStatAndMetric(proto.AdminGetCluster, metric, nil, nil)
+	}()
+
+	if err := r.ParseForm(); err != nil {
+		sendErrReply(w, r, newErrHTTPReply(err))
+		return
+	}
+	//TODO
+	cv := &proto.ClusterView{
+		//Name:                         m.cluster.Name,
+		//CreateTime:                   time.Unix(m.cluster.CreateTime, 0).Format(proto.TimeFormat),
+		//LeaderAddr:                   m.leaderInfo.addr,
+		//FlashNodes:                   make([]proto.NodeView, 0),
+		//FlashNodeHandleReadTimeout:   m.cluster.cfg.flashNodeHandleReadTimeout,
+		//FlashNodeReadDataNodeTimeout: m.cluster.cfg.flashNodeReadDataNodeTimeout,
+	}
+	cv.DataNodeStatInfo = new(proto.NodeStatInfo)
+	cv.MetaNodeStatInfo = new(proto.NodeStatInfo)
+	sendOkReply(w, r, newSuccessHTTPReply(cv))
+}
+
+func (m *FlashGroupMaster) getNodeInfoHandler(w http.ResponseWriter, r *http.Request) {
+	metric := exporter.NewTPCnt(apiToMetricsName(proto.AdminGetNodeInfo))
+	defer func() {
+		doStatAndMetric(proto.AdminGetNodeInfo, metric, nil, nil)
+	}()
+	//compatible for cli tool
+	resp := make(map[string]string)
+
+	sendOkReply(w, r, newSuccessHTTPReply(resp))
+}
+
+func (m *FlashGroupMaster) getIPAddr(w http.ResponseWriter, r *http.Request) {
+	metric := exporter.NewTPCnt(apiToMetricsName(proto.AdminGetIP))
+	defer func() {
+		doStatAndMetric(proto.AdminGetIP, metric, nil, nil)
+	}()
+	cInfo := &proto.ClusterInfo{
+		Cluster: m.clusterName,
+		Ip:      iputil.RealIP(r),
+	}
+	sendOkReply(w, r, newSuccessHTTPReply(cInfo))
+}

remotecache/flashgroupmaster/config.go

@@ -0,0 +1,19 @@
+package flashgroupmaster
+
+const (
+	defaultFlashNodeHandleReadTimeout   = 1000
+	defaultFlashNodeReadDataNodeTimeout = 3000
+)
+
+type clusterConfig struct {
+	flashNodeHandleReadTimeout   int
+	flashNodeReadDataNodeTimeout int
+}
+
+func newClusterConfig() (cfg *clusterConfig) {
+	cfg = new(clusterConfig)
+
+	cfg.flashNodeHandleReadTimeout = defaultFlashNodeHandleReadTimeout
+	cfg.flashNodeReadDataNodeTimeout = defaultFlashNodeReadDataNodeTimeout
+	return
+}

remotecache/flashgroupmaster/http_server.go

@@ -0,0 +1,120 @@
+package flashgroupmaster
+
+import (
+	"fmt"
+	"github.com/cubefs/cubefs/proto"
+	"net/http"
+	"time"
+
+	"github.com/cubefs/cubefs/util/config"
+	"github.com/cubefs/cubefs/util/exporter"
+	"github.com/cubefs/cubefs/util/log"
+	"github.com/gorilla/mux"
+)
+
+func (m *FlashGroupMaster) startHTTPService(modulename string, cfg *config.Config) {
+	router := mux.NewRouter().SkipClean(true)
+	m.registerAPIRoutes(router)
+	//m.registerAPIMiddleware(router)
+	exporter.InitWithRouter(modulename, cfg, router, m.port)
+	addr := fmt.Sprintf(":%s", m.port)
+	if m.bindIp {
+		addr = fmt.Sprintf("%s:%s", m.ip, m.port)
+	}
+
+	server := &http.Server{
+		Addr:         addr,
+		Handler:      router,
+		ReadTimeout:  5 * time.Minute,
+		WriteTimeout: 5 * time.Minute,
+	}
+
+	serveAPI := func() {
+		if err := server.ListenAndServe(); err != nil {
+			log.LogErrorf("serveAPI: serve http server failed: err(%v)", err)
+			return
+		}
+	}
+	go serveAPI()
+	m.apiServer = server
+}
+
+func (m *FlashGroupMaster) registerAPIRoutes(router *mux.Router) {
+	router.NewRoute().Methods(http.MethodGet).
+		Path(proto.AdminGetCluster).
+		HandlerFunc(m.getCluster)
+	router.NewRoute().Methods(http.MethodGet, http.MethodPost).
+		Path(proto.AdminGetNodeInfo).
+		HandlerFunc(m.getNodeInfoHandler)
+	router.NewRoute().Name(proto.AdminGetIP).
+		Methods(http.MethodGet).
+		Path(proto.AdminGetIP).
+		HandlerFunc(m.getIPAddr)
+}
+
+//TODO
+//func (m *FlashGroupMaster) registerAPIMiddleware(route *mux.Router) {
+//	var interceptor mux.MiddlewareFunc = func(next http.Handler) http.Handler {
+//		return http.HandlerFunc(
+//			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())
+//
+//				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 {
+//					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)
+//							errMsg := fmt.Sprintf("too many requests for api: %s", html.EscapeString(r.URL.Path))
+//							http.Error(w, errMsg, http.StatusTooManyRequests)
+//							return
+//						}
+//					}
+//				}
+//
+//				log.LogInfof("action[interceptor] request, remote[%v] method[%v] path[%v] query[%v]",
+//					r.RemoteAddr, r.Method, r.URL.Path, r.URL.Query())
+//				if mux.CurrentRoute(r).GetName() == proto.AdminGetIP {
+//					next.ServeHTTP(w, r)
+//					return
+//				}
+//
+//				isFollowerRead := m.isFollowerRead(r)
+//				if m.partition.IsRaftLeader() || isFollowerRead {
+//					if m.metaReady || isFollowerRead {
+//						log.LogDebugf("action[interceptor] request, method[%v] path[%v] query[%v]", r.Method, r.URL.Path, r.URL.Query())
+//						next.ServeHTTP(w, r)
+//						return
+//					}
+//					log.LogWarnf("action[interceptor] leader meta has not ready")
+//					http.Error(w, m.leaderInfo.addr, http.StatusBadRequest)
+//					return
+//				} else if m.leaderInfo.addr != "" {
+//					if m.isClientPartitionsReq(r) && m.cluster.cfg.EnableFollowerCache {
+//						log.LogErrorf("action[interceptor] request, method[%v] path[%v] query[%v] status [%v]", r.Method, r.URL.Path, r.URL.Query(), isFollowerRead)
+//						http.Error(w, m.leaderInfo.addr, http.StatusBadRequest)
+//						return
+//					}
+//
+//					if m.leaderInfo.addr == m.getCurrAddr() {
+//						log.LogErrorf("action[interceptor] request, self is leader addr, no leader now, method[%v] path[%v] query[%v] status [%v]",
+//							r.Method, r.URL.Path, r.URL.Query(), m.leaderInfo.addr)
+//						http.Error(w, "no leader", http.StatusBadRequest)
+//						return
+//					}
+//
+//					m.proxy(w, r)
+//				} else {
+//					log.LogErrorf("action[interceptor] no leader,request[%v]", r.URL)
+//					http.Error(w, "no leader", http.StatusBadRequest)
+//					return
+//				}
+//			})
+//	}
+//	route.Use(interceptor)
+//}

remotecache/flashgroupmaster/server.go

@@ -0,0 +1,98 @@
+package flashgroupmaster
+
+import (
+	"fmt"
+	"net/http"
+	"sync"
+
+	"github.com/cubefs/cubefs/proto"
+	"github.com/cubefs/cubefs/util/config"
+	"github.com/cubefs/cubefs/util/errors"
+	"github.com/cubefs/cubefs/util/exporter"
+	"github.com/cubefs/cubefs/util/log"
+)
+
+const (
+	ModuleName = "flashGroupMaster"
+)
+
+const (
+	ClusterName = "clusterName"
+	IP          = "ip"
+)
+
+type FlashGroupMaster struct {
+	wg          sync.WaitGroup
+	clusterName string
+	config      *clusterConfig
+	ip          string
+	bindIp      bool
+	port        string
+	apiServer   *http.Server
+}
+
+func NewFlashGroupMaster() *FlashGroupMaster {
+	return &FlashGroupMaster{}
+}
+
+func (m *FlashGroupMaster) Start(cfg *config.Config) (err error) {
+	m.config = newClusterConfig()
+	if err = m.checkConfig(cfg); err != nil {
+		log.LogError(errors.Stack(err))
+		return
+	}
+
+	// TODO
+	// m.leaderInfo = &LeaderInfo{}
+
+	// TODO
+	// if m.rocksDBStore, err = raftstore_db.NewRocksDBStoreAndRecovery(m.storeDir, LRUCacheSize, WriteBufferSize); err != nil {
+	// TODO
+	// if err = m.createRaftServer(cfg); err != nil {
+
+	// TODO
+	// initCluster
+
+	// TODO:need?
+	// m.cluster.initAuthentication(cfg)
+
+	// TODO
+	// WarnMetrics = newWarningMetrics(m.cluster)
+
+	// TODO
+	// scheduleTask
+
+	m.startHTTPService(ModuleName, cfg)
+	exporter.RegistConsul(m.clusterName, ModuleName, cfg)
+	exporter.SetBuckets([]float64{1000})
+	// TODO
+	// newMonitorMetrics
+
+	// TODO
+	// stat.NewStatistic
+
+	m.wg.Add(1)
+	return
+}
+
+func (m *FlashGroupMaster) Shutdown() {
+
+	m.wg.Done()
+}
+
+func (m *FlashGroupMaster) Sync() {
+	m.wg.Wait()
+}
+
+func (m *FlashGroupMaster) checkConfig(cfg *config.Config) (err error) {
+	m.clusterName = cfg.GetString(ClusterName)
+	m.ip = cfg.GetString(IP)
+	m.bindIp = cfg.GetBool(proto.BindIpKey)
+	m.port = cfg.GetString(proto.ListenPort)
+
+	if m.port == "" || m.clusterName == "" {
+		return fmt.Errorf("%v,err:%v,%v,%v", proto.ErrInvalidCfg, "one of (listen,walDir,clusterName) is null",
+			m.ip, m.clusterName)
+	}
+	return
+}