From a51f298c797fd8c476fc1759795587917a4326d1 Mon Sep 17 00:00:00 2001 From: AI Engineer Date: Tue, 5 May 2026 09:42:15 +0800 Subject: [PATCH] feat: migrate discover module from ssgo to apigo.cc/go standard (by AI) --- AppClient.go | 90 ++++++++++ CHANGELOG.md | 9 + Caller.go | 242 +++++++++++++++++++++++++ Config.go | 14 ++ Constants.go | 38 ++++ Discover.go | 449 +++++++++++++++++++++++++++++++++++++++++++++++ Discover_test.go | 140 +++++++++++++++ LoadBalancer.go | 45 +++++ NodeInfo.go | 14 ++ README.md | 48 +++++ Route.go | 8 + TEST.md | 15 ++ go.mod | 29 +++ go.sum | 48 +++++ 14 files changed, 1189 insertions(+) create mode 100644 AppClient.go create mode 100644 CHANGELOG.md create mode 100644 Caller.go create mode 100644 Config.go create mode 100644 Constants.go create mode 100644 Discover.go create mode 100644 Discover_test.go create mode 100644 LoadBalancer.go create mode 100644 NodeInfo.go create mode 100644 README.md create mode 100644 Route.go create mode 100644 TEST.md create mode 100644 go.mod create mode 100644 go.sum diff --git a/AppClient.go b/AppClient.go new file mode 100644 index 0000000..c8c8891 --- /dev/null +++ b/AppClient.go @@ -0,0 +1,90 @@ +package discover + +import ( + "net/http" + + "apigo.cc/go/log" +) + +// AppClient 用于管理单个请求的重试和负载均衡状态 +type AppClient struct { + excludes map[string]bool + tryTimes int + Logger *log.Logger + App string + Method string + Path string + Data *map[string]any + Headers *map[string]string +} + +func (ac *AppClient) logError(error string, extra ...any) { + if ac.Logger == nil { + ac.Logger = log.DefaultLogger + } + ac.Logger.Error("Discover Client: "+error, extra...) +} + +func (ac *AppClient) Next(app string, request *http.Request) *NodeInfo { + return ac.NextWithNode(app, "", request) +} + +func (ac *AppClient) CheckApp(app string) bool { + nodes := getAppNodes(app) + if nodes == nil { + if !addApp(app, "", true) { + ac.logError("app not found", "app", app, "calls", Config.Calls) + return false + } + } + return true +} + +func (ac *AppClient) NextWithNode(app, withNode string, request *http.Request) *NodeInfo { + if ac.excludes == nil { + ac.excludes = make(map[string]bool) + } + + allNodes := getAppNodes(app) + if len(allNodes) == 0 { + ac.logError("node not found", "app", app) + return nil + } + + ac.tryTimes++ + if withNode != "" { + ac.excludes[withNode] = true + return allNodes[withNode] + } + + readyNodes := make([]*NodeInfo, 0) + for _, node := range allNodes { + if ac.excludes[node.Addr] || node.FailedTimes >= Config.CallRetryTimes { + continue + } + readyNodes = append(readyNodes, node) + } + + if len(readyNodes) == 0 { + // 如果没有可用节点，尝试已经失败但未被本次请求排除的节点 + for _, node := range allNodes { + if !ac.excludes[node.Addr] { + readyNodes = append(readyNodes, node) + } + } + } + + var node *NodeInfo + if len(readyNodes) > 0 { + node = settedLoadBalancer.Next(ac, readyNodes, request) + if node != nil { + ac.excludes[node.Addr] = true + } + } + + if node == nil { + ac.logError("no available node", "app", app, "tryTimes", ac.tryTimes) + } + + return node +} diff --git a/CHANGELOG.md b/CHANGELOG.md new file mode 100644 index 0000000..a27776f --- /dev/null +++ b/CHANGELOG.md @@ -0,0 +1,9 @@ +# CHANGELOG + +## v1.0.0 +- 从 `ssgo/discover` 迁移至 `apigo.cc/go/discover`。 +- 采用全新的 `apigo.cc/go` 基础设施（log, redis, http, cast, u）。 +- 优化了注册中心同步机制，使用 `redis.Subscribe` 简化 PubSub 处理。 +- 增强了负载均衡算法，引入更精确的得分计算。 +- 统一了 Header 定义，对齐 `go/http` 标准。 +- 移除所有 `panic`，通过 `error` 返回和日志记录确保系统稳定性。 diff --git a/Caller.go b/Caller.go new file mode 100644 index 0000000..87dacbf --- /dev/null +++ b/Caller.go @@ -0,0 +1,242 @@ +package discover + +import ( + "fmt" + "net/http" + "reflect" + "strings" + "sync" + "time" + + "github.com/gorilla/websocket" + "apigo.cc/go/cast" + gohttp "apigo.cc/go/http" + "apigo.cc/go/log" +) + +var appClientPools = make(map[string]*gohttp.Client) +var appClientPoolsLock sync.RWMutex + +func getHttpClient(app string, timeout time.Duration, h2c bool) *gohttp.Client { + appClientPoolsLock.RLock() + c := appClientPools[app] + appClientPoolsLock.RUnlock() + if c != nil { + return c + } + + appClientPoolsLock.Lock() + defer appClientPoolsLock.Unlock() + c = appClientPools[app] + if c != nil { + return c + } + + if h2c { + c = gohttp.NewClientH2C(timeout) + } else { + c = gohttp.NewClient(timeout) + } + appClientPools[app] = c + return c +} + +type Caller struct { + Request *http.Request + NoBody bool + logger *log.Logger +} + +func NewCaller(request *http.Request, logger *log.Logger) *Caller { + return &Caller{Request: request, logger: logger} +} + +func (c *Caller) logError(error string, extra ...any) { + if c.logger == nil { + c.logger = log.DefaultLogger + } + c.logger.Error("Discover Caller: "+error, extra...) +} + +func (c *Caller) Get(app, path string, headers ...string) *gohttp.Result { + return c.Do("GET", app, path, nil, headers...) +} + +func (c *Caller) Post(app, path string, data any, headers ...string) *gohttp.Result { + return c.Do("POST", app, path, data, headers...) +} + +func (c *Caller) Put(app, path string, data any, headers ...string) *gohttp.Result { + return c.Do("PUT", app, path, data, headers...) +} + +func (c *Caller) Delete(app, path string, data any, headers ...string) *gohttp.Result { + return c.Do("DELETE", app, path, data, headers...) +} + +func (c *Caller) Head(app, path string, headers ...string) *gohttp.Result { + return c.Do("HEAD", app, path, nil, headers...) +} + +func (c *Caller) Do(method, app, path string, data any, headers ...string) *gohttp.Result { + r, _ := c.DoWithNode(method, app, "", path, data, headers...) + return r +} + +func (c *Caller) Open(app, path string, headers ...string) *websocket.Conn { + r, _ := c.doWithNode(false, "WS", app, "", path, nil, headers...) + if v, ok := r.(*websocket.Conn); ok { + return v + } + return nil +} + +func (c *Caller) DoWithNode(method, app, withNode, path string, data any, headers ...string) (*gohttp.Result, string) { + r, nodeAddr := c.doWithNode(false, method, app, withNode, path, data, headers...) + if v, ok := r.(*gohttp.Result); ok { + return v, nodeAddr + } + return nil, nodeAddr +} + +func (c *Caller) ManualDoWithNode(method, app, withNode, path string, data any, headers ...string) (*gohttp.Result, string) { + r, nodeAddr := c.doWithNode(true, method, app, withNode, path, data, headers...) + if v, ok := r.(*gohttp.Result); ok { + return v, nodeAddr + } + return nil, nodeAddr +} + +func (c *Caller) doWithNode(manualDo bool, method, app, withNode, path string, data any, headers ...string) (any, string) { + callerHeaders := make(map[string]string) + for i := 1; i < len(headers); i += 2 { + callerHeaders[headers[i-1]] = headers[i] + } + + if isServer { + callerHeaders[HeaderFromApp] = Config.App + callerHeaders[HeaderFromNode] = myAddr + } + + callData := make(map[string]any) + if data != nil && !c.NoBody { + rv := cast.RealValue(reflect.ValueOf(data)) + if rv.Kind() == reflect.Map || rv.Kind() == reflect.Struct { + cast.Convert(&callData, data) + } + } + + appClient := AppClient{ + Logger: c.logger, + App: app, + Method: method, + Path: path, + Data: &callData, + Headers: &callerHeaders, + } + + if settedRoute != nil { + settedRoute(&appClient, c.Request) + app = appClient.App + method = appClient.Method + path = appClient.Path + } + + if !appClient.CheckApp(app) { + return &gohttp.Result{Error: fmt.Errorf("app %s not found", app)}, "" + } + + callInfo := getCallInfo(app) + if callInfo != nil && callInfo.Token != "" { + callerHeaders["Access-Token"] = callInfo.Token + } + + settedHeaders := make([]string, 0, len(callerHeaders)*2) + for k, v := range callerHeaders { + settedHeaders = append(settedHeaders, k, v) + } + + for { + node := appClient.NextWithNode(app, withNode, c.Request) + if node == nil { + break + } + + node.UsedTimes++ + startTime := time.Now() + scheme := "http" + if callInfo != nil && callInfo.SSL { + scheme = "https" + } + + hc := getHttpClient(app, callInfo.Timeout, callInfo.HttpVersion == 2 && !callInfo.SSL) + hc.NoBody = c.NoBody + + var res *gohttp.Result + var wsConn *websocket.Conn + + url := fmt.Sprintf("%s://%s%s", scheme, node.Addr, path) + + if strings.ToUpper(method) == "WS" { + dialer := websocket.DefaultDialer + h := http.Header{} + for i := 1; i < len(settedHeaders); i += 2 { + h.Set(settedHeaders[i-1], settedHeaders[i]) + } + if scheme == "https" { + scheme = "wss" + } else { + scheme = "ws" + } + wsUrl := fmt.Sprintf("%s://%s%s", scheme, node.Addr, path) + conn, resp, err := dialer.Dial(wsUrl, h) + wsConn = conn + res = &gohttp.Result{Error: err, Response: resp} + } else { + if c.Request != nil { + if manualDo { + res = hc.ManualDoByRequest(c.Request, method, url, data, settedHeaders...) + } else { + res = hc.DoByRequest(c.Request, method, url, data, settedHeaders...) + } + } else { + if manualDo { + res = hc.ManualDo(method, url, data, settedHeaders...) + } else { + res = hc.Do(method, url, data, settedHeaders...) + } + } + } + + responseTime := time.Since(startTime) + settedLoadBalancer.Response(&appClient, node, res.Error, res.Response, responseTime) + + if res.Error != nil || (res.Response != nil && res.Response.StatusCode >= 502 && res.Response.StatusCode <= 504) { + node.FailedTimes++ + errStr := "" + if res.Error != nil { + errStr = res.Error.Error() + } else { + errStr = res.Response.Status + } + + c.logError(errStr, "app", app, "node", node.Addr, "path", path, "tryTimes", appClient.tryTimes) + + if node.FailedTimes >= Config.CallRetryTimes { + logError("node removed due to high failures", "app", app, "node", node.Addr) + if clientRedisPool != nil { + clientRedisPool.Do("HDEL", app, node.Addr) + clientRedisPool.PUBLISH("CH_"+app, fmt.Sprintf("%s 0", node.Addr)) + } + } + continue + } + + if strings.ToUpper(method) == "WS" { + return wsConn, node.Addr + } + return res, node.Addr + } + + return &gohttp.Result{Error: fmt.Errorf("all nodes failed for %s %s", app, path)}, "" +} diff --git a/Config.go b/Config.go new file mode 100644 index 0000000..3c4bfbf --- /dev/null +++ b/Config.go @@ -0,0 +1,14 @@ +package discover + +// Config 存储发现服务的全局配置 +var Config = struct { + Registry string // 注册中心地址，如 redis://:@127.0.0.1:6379/15 + App string // 当前应用名称 + Weight int // 权重，默认为 100 + Calls map[string]string // 调用的应用列表及其配置 + CallRetryTimes int // 调用重试次数 + IpPrefix string // 指定使用的 IP 网段 +}{ + Weight: 100, + CallRetryTimes: 10, +} diff --git a/Constants.go b/Constants.go new file mode 100644 index 0000000..633a7b9 --- /dev/null +++ b/Constants.go @@ -0,0 +1,38 @@ +package discover + +const ( + HeaderFromApp = "X-Discover-From-App" + HeaderFromNode = "X-Discover-From-Node" + + HeaderClientIp = "X-Client-Ip" + HeaderForwardedFor = "X-Forwarded-For" + HeaderUserId = "X-User-Id" + HeaderDeviceId = "X-Device-Id" + HeaderClientAppName = "X-Client-App-Name" + HeaderClientAppVersion = "X-Client-App-Version" + HeaderSessionId = "X-Session-Id" + HeaderRequestId = "X-Request-Id" + HeaderHost = "X-Host" + HeaderScheme = "X-Scheme" + HeaderUserAgent = "User-Agent" +) + +var RelayHeaders = []string{ + HeaderClientIp, + HeaderForwardedFor, + HeaderUserId, + HeaderDeviceId, + HeaderClientAppName, + HeaderClientAppVersion, + HeaderSessionId, + HeaderRequestId, + HeaderHost, + HeaderScheme, + HeaderUserAgent, +} + +const DefaultRegistry = "127.0.0.1:6379::15" +const EnvRegistry = "DISCOVER_REGISTRY" +const EnvApp = "DISCOVER_APP" +const EnvWeight = "DISCOVER_WEIGHT" +const EnvCalls = "DISCOVER_CALLS" diff --git a/Discover.go b/Discover.go new file mode 100644 index 0000000..c629259 --- /dev/null +++ b/Discover.go @@ -0,0 +1,449 @@ +package discover + +import ( + "fmt" + "net" + "net/http" + "os" + "os/signal" + "regexp" + "strings" + "sync" + "syscall" + "time" + + "apigo.cc/go/cast" + "apigo.cc/go/config" + "apigo.cc/go/id" + "apigo.cc/go/log" + "apigo.cc/go/redis" +) + +var ( + serverRedisPool *redis.Redis + clientRedisPool *redis.Redis + pubsubRedisPool *redis.Redis + isServer = false + isClient = false + daemonRunning = false + myAddr = "" + _logger = log.DefaultLogger + _inited = false + + daemonStopChan chan bool + appLock sync.RWMutex + _calls = map[string]*callInfoType{} + _appNodes = map[string]map[string]*NodeInfo{} + appSubscribed = map[string]bool{} + + settedRoute func(*AppClient, *http.Request) = nil + settedLoadBalancer LoadBalancer = &DefaultLoadBalancer{} +) + +type callInfoType struct { + Timeout time.Duration + HttpVersion int + Token string + SSL bool +} + +func IsServer() bool { return isServer } +func IsClient() bool { return isClient } + +func logError(error string, extra ...any) { + _logger.Error("Discover: "+error, append(extra, "app", Config.App, "addr", myAddr)...) +} + +func logInfo(info string, extra ...any) { + _logger.Info("Discover: "+info, append(extra, "app", Config.App, "addr", myAddr)...) +} + +func SetLogger(logger *log.Logger) { + _logger = logger +} + +func Init() { + appLock.Lock() + defer appLock.Unlock() + if _inited { + return + } + _inited = true + _ = config.Load(&Config, "discover") + + if Config.CallRetryTimes <= 0 { + Config.CallRetryTimes = 10 + } + if Config.Weight <= 0 { + Config.Weight = 100 + } + if Config.Registry == "" { + Config.Registry = DefaultRegistry + } + + _logger = log.New(id.MakeID(12)) +} + +func Start(addr string) bool { + Init() + myAddr = addr + + isServer = Config.App != "" && Config.Weight > 0 + if isServer && Config.Registry != "" { + serverRedisPool = redis.GetRedis(Config.Registry, _logger) + if serverRedisPool.Error != nil { + logError(serverRedisPool.Error.Error()) + } + + // 注册节点 + if serverRedisPool.Do("HSET", Config.App, addr, Config.Weight).Error == nil { + serverRedisPool.Do("SETEX", Config.App+"_"+addr, 10, "1") + logInfo("registered") + serverRedisPool.PUBLISH("CH_"+Config.App, fmt.Sprintf("%s %d", addr, Config.Weight)) + daemonRunning = true + daemonStopChan = make(chan bool) + go daemon() + } else { + logError("register failed") + } + } + + calls := getCalls() + if len(calls) > 0 { + for app, conf := range calls { + addApp(app, conf, false) + } + if !startSub() { + return false + } + } + return true +} + +func daemon() { + logInfo("daemon thread started") + ticker := time.NewTicker(time.Second) + defer ticker.Stop() + + for daemonRunning { + <-ticker.C + if !daemonRunning { + break + } + + if isServer && serverRedisPool != nil { + if !serverRedisPool.Do("HEXISTS", Config.App, myAddr).Bool() { + logInfo("lost app registered info, re-registering") + if serverRedisPool.Do("HSET", Config.App, myAddr, Config.Weight).Error == nil { + serverRedisPool.Do("SETEX", Config.App+"_"+myAddr, 10, "1") + serverRedisPool.PUBLISH("CH_"+Config.App, fmt.Sprintf("%s %d", myAddr, Config.Weight)) + } + } else { + serverRedisPool.Do("SETEX", Config.App+"_"+myAddr, 10, "1") + } + } + } + logInfo("daemon thread stopped") + if daemonStopChan != nil { + daemonStopChan <- true + } +} + +func startSub() bool { + if Config.Registry == "" { + return true + } + + appLock.Lock() + if clientRedisPool == nil { + clientRedisPool = redis.GetRedis(Config.Registry, _logger) + } + + if pubsubRedisPool == nil { + pubsubRedisPool = redis.GetRedis(Config.Registry, _logger.New(id.MakeID(12))) + // 订阅所有已注册的应用 + for app := range appSubscribed { + subscribeAppUnderLock(app) + } + // 必须在释放锁之前完成配置，但在释放锁之后启动，避免死锁 + appLock.Unlock() + pubsubRedisPool.Start() + appLock.Lock() + } + + isClient = true + appLock.Unlock() + return true +} + +func subscribeAppUnderLock(app string) { + pubsubRedisPool.Subscribe("CH_"+app, func() { + fetchApp(app) + }, func(data []byte) { + a := strings.Split(string(data), " ") + addr := a[0] + weight := 0 + if len(a) == 2 { + weight = cast.Int(a[1]) + } + logInfo("received node update", "app", app, "addr", addr, "weight", weight) + pushNode(app, addr, weight) + }) +} + +func Stop() { + appLock.Lock() + if isClient && pubsubRedisPool != nil { + pubsubRedisPool.Stop() + isClient = false + } + + if isServer { + daemonRunning = false + if serverRedisPool != nil { + serverRedisPool.Do("HDEL", Config.App, myAddr) + serverRedisPool.Do("DEL", Config.App+"_"+myAddr) + serverRedisPool.PUBLISH("CH_"+Config.App, fmt.Sprintf("%s %d", myAddr, 0)) + } + isServer = false + } + appLock.Unlock() +} + +func Wait() { + if daemonStopChan != nil { + <-daemonStopChan + daemonStopChan = nil + } +} + +func EasyStart() (string, int) { + Init() + port := 0 + if listen := os.Getenv("DISCOVER_LISTEN"); listen != "" { + if _, p, err := net.SplitHostPort(listen); err == nil { + port = cast.Int(p) + } else { + port = cast.Int(listen) + } + } + + ln, err := net.Listen("tcp", fmt.Sprintf(":%d", port)) + if err != nil { + logError("failed to listen", "err", err) + return "", 0 + } + addrInfo := ln.Addr().(*net.TCPAddr) + _ = ln.Close() + port = addrInfo.Port + + ip := addrInfo.IP + if !ip.IsGlobalUnicast() { + addrs, _ := net.InterfaceAddrs() + for _, a := range addrs { + if an, ok := a.(*net.IPNet); ok { + ip4 := an.IP.To4() + if ip4 == nil || !ip4.IsGlobalUnicast() { + continue + } + if Config.IpPrefix != "" && strings.HasPrefix(ip4.String(), Config.IpPrefix) { + ip = ip4 + break + } + if !strings.HasPrefix(ip4.String(), "172.17.") { + ip = ip4 + } + } + } + } + + addr := fmt.Sprintf("%s:%d", ip.String(), port) + if !Start(addr) { + return "", 0 + } + + sigChan := make(chan os.Signal, 1) + signal.Notify(sigChan, os.Interrupt, syscall.SIGTERM) + go func() { + <-sigChan + Stop() + }() + + return ip.String(), port +} + +func AddExternalApp(app, callConf string) bool { + if addApp(app, callConf, true) { + if !isClient { + startSub() + } else { + appLock.Lock() + subscribeAppUnderLock(app) + appLock.Unlock() + } + return true + } + return false +} + +func SetNode(app, addr string, weight int) { + pushNode(app, addr, weight) +} + +func getCallInfo(app string) *callInfoType { + appLock.RLock() + defer appLock.RUnlock() + return _calls[app] +} + +var numberMatcher = regexp.MustCompile(`^\d+(s|ms|us|µs|ns?)?$`) + +func addApp(app, callConf string, fetch bool) bool { + appLock.Lock() + if Config.Calls == nil { + Config.Calls = make(map[string]string) + } + if Config.Calls[app] == callConf && _appNodes[app] != nil { + appLock.Unlock() + return false + } + Config.Calls[app] = callConf + + callInfo := &callInfoType{ + Timeout: 10 * time.Second, + HttpVersion: 2, + SSL: false, + } + + for _, v := range cast.Split(callConf, ":") { + switch v { + case "1": + callInfo.HttpVersion = 1 + case "2": + callInfo.HttpVersion = 2 + case "s", "https": + callInfo.SSL = true + callInfo.HttpVersion = 2 + case "http": + callInfo.SSL = false + callInfo.HttpVersion = 1 + case "h2c": + callInfo.SSL = false + callInfo.HttpVersion = 2 + default: + if numberMatcher.MatchString(v) { + callInfo.Timeout = cast.Duration(v) + } else { + callInfo.Token = v + } + } + } + + _calls[app] = callInfo + if _appNodes[app] == nil { + _appNodes[app] = make(map[string]*NodeInfo) + } + appSubscribed[app] = true + appLock.Unlock() + + if fetch { + fetchApp(app) + } + return true +} + +func fetchApp(app string) { + appLock.RLock() + pool := clientRedisPool + appLock.RUnlock() + if pool == nil { + return + } + + results := pool.Do("HGETALL", app).ResultMap() + + // 检查存活 + for addr := range results { + if !pool.Do("EXISTS", app+"_"+addr).Bool() { + pool.Do("HDEL", app, addr) + delete(results, addr) + } + } + + currentNodes := getAppNodes(app) + if currentNodes != nil { + for addr := range currentNodes { + if _, ok := results[addr]; !ok { + pushNode(app, addr, 0) + } + } + } + + for addr, res := range results { + pushNode(app, addr, res.Int()) + } +} + +func getAppNodes(app string) map[string]*NodeInfo { + appLock.RLock() + defer appLock.RUnlock() + if _appNodes[app] == nil { + return nil + } + nodes := make(map[string]*NodeInfo) + for k, v := range _appNodes[app] { + nodes[k] = v + } + return nodes +} + +func getCalls() map[string]string { + appLock.RLock() + defer appLock.RUnlock() + calls := make(map[string]string) + for k, v := range Config.Calls { + calls[k] = v + } + return calls +} + +func GetAppNodes(app string) map[string]*NodeInfo { + return getAppNodes(app) +} + +func pushNode(app, addr string, weight int) { + appLock.Lock() + defer appLock.Unlock() + + if weight <= 0 { + if _appNodes[app] != nil { + delete(_appNodes[app], addr) + } + return + } + + if _appNodes[app] == nil { + _appNodes[app] = make(map[string]*NodeInfo) + } + + if node, ok := _appNodes[app][addr]; ok { + if node.Weight != weight { + node.UsedTimes = uint64(float64(node.UsedTimes) / float64(node.Weight) * float64(weight)) + node.Weight = weight + } + } else { + var avgUsed uint64 = 0 + if len(_appNodes[app]) > 0 { + var totalScore float64 + for _, n := range _appNodes[app] { + totalScore += float64(n.UsedTimes) / float64(n.Weight) + } + avgUsed = uint64(totalScore / float64(len(_appNodes[app])) * float64(weight)) + } + _appNodes[app][addr] = &NodeInfo{ + Addr: addr, + Weight: weight, + UsedTimes: avgUsed, + } + } +} diff --git a/Discover_test.go b/Discover_test.go new file mode 100644 index 0000000..6d058ad --- /dev/null +++ b/Discover_test.go @@ -0,0 +1,140 @@ +package discover_test + +import ( + "fmt" + "net" + "net/http" + "os" + "testing" + "time" + + "github.com/gorilla/websocket" + "apigo.cc/go/discover" + "apigo.cc/go/redis" +) + +func TestDiscover(t *testing.T) { + // 启动一个模拟服务 + l, err := net.Listen("tcp", "127.0.0.1:18001") + if err != nil { + t.Skip("failed to listen on :18001, skipping test") + return + } + mux := http.NewServeMux() + mux.HandleFunc("/", func(w http.ResponseWriter, r *http.Request) { + _, _ = w.Write([]byte("OK")) + }) + + upgrader := websocket.Upgrader{} + mux.HandleFunc("/ws", func(w http.ResponseWriter, r *http.Request) { + conn, err := upgrader.Upgrade(w, r, nil) + if err != nil { + return + } + defer conn.Close() + for { + mt, message, err := conn.ReadMessage() + if err != nil { + break + } + _ = conn.WriteMessage(mt, message) + } + }) + + server := &http.Server{Handler: mux} + go func() { _ = server.Serve(l) }() + defer server.Close() + + // 配置 Discover + discover.Config.App = "test-app" + discover.Config.Registry = "redis://127.0.0.1:6379/15" + + // 启动 Discover + if !discover.Start("127.0.0.1:18001") { + t.Skip("failed to start discover (check redis), skipping test") + return + } + defer discover.Stop() + + // 添加外部应用调用配置 + discover.AddExternalApp("test-app", "1") + + // 等待节点同步 + success := false + for i := 0; i < 20; i++ { + nodes := discover.GetAppNodes("test-app") + if len(nodes) > 0 { + success = true + break + } + time.Sleep(100 * time.Millisecond) + } + if !success { + t.Fatal("node discovery timed out") + } + + // 1. 使用 Caller 调用 HTTP + caller := discover.NewCaller(nil, nil) + res := caller.Get("test-app", "/") + if res.Error != nil { + t.Errorf("http call failed: %v", res.Error) + } + if res.String() != "OK" { + t.Errorf("unexpected http response: %s", res.String()) + } + + // 2. 使用 Caller 调用 WebSocket + wsConn := caller.Open("test-app", "/ws") + if wsConn == nil { + t.Fatal("websocket open failed") + } + defer wsConn.Close() + + msg := []byte("hello") + if err := wsConn.WriteMessage(websocket.TextMessage, msg); err != nil { + t.Fatalf("ws write failed: %v", err) + } + + _, reply, err := wsConn.ReadMessage() + if err != nil { + t.Fatalf("ws read failed: %v", err) + } + if string(reply) != "hello" { + t.Errorf("unexpected ws reply: %s", string(reply)) + } + + // 3. 测试负载均衡和节点更新 + rd := redis.GetRedis(discover.Config.Registry, nil) + if rd.Error == nil { + // 模拟发现新节点 + rd.PUBLISH("CH_test-app", "127.0.0.1:18002 100") + + success = false + for i := 0; i < 20; i++ { + nodes := discover.GetAppNodes("test-app") + if len(nodes) >= 2 { + success = true + break + } + time.Sleep(100 * time.Millisecond) + } + if !success { + t.Error("node update sync failed") + } + } +} + +func TestEasyStart(t *testing.T) { + // 模拟环境变量 + _ = os.Setenv("DISCOVER_APP", "test-app") + _ = os.Setenv("DISCOVER_LISTEN", "18003") + _ = os.Setenv("DISCOVER_REGISTRY", "redis://127.0.0.1:6379/15") + + ip, port := discover.EasyStart() + if ip == "" || port == 0 { + t.Skip("EasyStart failed (check redis), skipping test") + return + } + fmt.Printf("EasyStart: %s:%d\n", ip, port) + discover.Stop() +} diff --git a/LoadBalancer.go b/LoadBalancer.go new file mode 100644 index 0000000..2aa062b --- /dev/null +++ b/LoadBalancer.go @@ -0,0 +1,45 @@ +package discover + +import ( + "net/http" + "time" +) + +// SetLoadBalancer 设置全局负载均衡策略 +func SetLoadBalancer(lb LoadBalancer) { + settedLoadBalancer = lb +} + +// LoadBalancer 负载均衡接口 +type LoadBalancer interface { + // Response 在每个请求完成后调用，用于更新节点状态 + Response(appClient *AppClient, node *NodeInfo, err error, response *http.Response, responseTime time.Duration) + + // Next 根据当前可用节点选择一个最优节点 + Next(appClient *AppClient, nodes []*NodeInfo, request *http.Request) *NodeInfo +} + +// DefaultLoadBalancer 默认负载均衡器（简单权重轮询/得分最小者优先） +type DefaultLoadBalancer struct{} + +func (lb *DefaultLoadBalancer) Response(appClient *AppClient, node *NodeInfo, err error, response *http.Response, responseTime time.Duration) { + node.Data.Store("score", float64(node.UsedTimes)/float64(node.Weight)) +} + +func (lb *DefaultLoadBalancer) Next(appClient *AppClient, nodes []*NodeInfo, request *http.Request) *NodeInfo { + var minScore float64 = -1 + var minNode *NodeInfo + for _, node := range nodes { + scoreValue, ok := node.Data.Load("score") + if !ok { + scoreValue = float64(node.UsedTimes) / float64(node.Weight) + node.Data.Store("score", scoreValue) + } + score := scoreValue.(float64) + if minNode == nil || score < minScore { + minScore = score + minNode = node + } + } + return minNode +} diff --git a/NodeInfo.go b/NodeInfo.go new file mode 100644 index 0000000..941fde7 --- /dev/null +++ b/NodeInfo.go @@ -0,0 +1,14 @@ +package discover + +import ( + "sync" +) + +// NodeInfo 存储服务节点信息 +type NodeInfo struct { + Addr string // 节点地址 + Weight int // 节点权重 + UsedTimes uint64 // 已使用次数 + FailedTimes int // 失败次数 + Data sync.Map // 运行时自定义数据 +} diff --git a/README.md b/README.md new file mode 100644 index 0000000..a29bc56 --- /dev/null +++ b/README.md @@ -0,0 +1,48 @@ +# Discover + +基于 Redis 的极简服务发现与负载均衡组件。 + +## 核心特性 +- **自动注册与发现**: 基于 Redis 的服务节点自动注册、心跳维持及实时更新。 +- **智能负载均衡**: 支持按权重分配、自动剔除故障节点、重试机制。 +- **无感透传**: 自动处理微服务间的 Header 透传（如 TraceID、UserID 等）。 +- **多协议支持**: 支持 HTTP/1.1、HTTP/2 (H2C)、WebSocket。 + +## 配置参考 +```yaml +discover: + registry: redis://127.0.0.1:6379/15 # 注册中心地址 + app: my-service # 当前应用名称 + weight: 100 # 节点权重 + calls: # 调用的服务定义 + auth: 1s:my-token:2 # 服务名: 超时:Token:HTTP版本 + user: 500ms +``` + +## API 指南 + +### 初始化与启动 +- `Start(addr string) bool`: 启动服务发现，指定当前节点的外部访问地址。 +- `EasyStart() (string, int)`: 自动监听可用端口并启动服务发现。返回 IP 和端口。 +- `Stop()`: 停止服务并注销节点。 + +### 服务调用 (Caller) +- `NewCaller(request *http.Request, logger *log.Logger) *Caller`: 创建调用器。传入原始请求可自动透传 Header。 +- `Caller.Get / Post / Put / Delete / Head`: 发起同步请求。 +- `Caller.Do(method, app, path, data, headers...)`: 发起通用请求，返回 `http.Result`。 +- `Caller.Open(app, path, headers...)`: 发起 WebSocket 连接。 + +### 手动管理 +- `AddExternalApp(app, callConf string)`: 手动添加需要发现的外部应用。 +- `SetNode(app, addr string, weight int)`: 手动设置某个服务的节点信息。 + +### 负载均衡与路由 +- `SetLoadBalancer(lb LoadBalancer)`: 自定义全局负载均衡策略。 +- `SetRoute(route func(ac *AppClient, r *http.Request))`: 设置全局路由拦截规则。 + +## 环境变量 +- `DISCOVER_REGISTRY`: 注册中心地址。 +- `DISCOVER_APP`: 应用名。 +- `DISCOVER_WEIGHT`: 节点权重。 +- `DISCOVER_CALLS`: 调用的应用定义。 +- `DISCOVER_LISTEN`: EasyStart 监听地址。 diff --git a/Route.go b/Route.go new file mode 100644 index 0000000..c339f78 --- /dev/null +++ b/Route.go @@ -0,0 +1,8 @@ +package discover + +import "net/http" + +// SetRoute 设置全局路由规则，可以在请求前修改 App、Method、Path 等信息 +func SetRoute(route func(appClient *AppClient, request *http.Request)) { + settedRoute = route +} diff --git a/TEST.md b/TEST.md new file mode 100644 index 0000000..440f1b7 --- /dev/null +++ b/TEST.md @@ -0,0 +1,15 @@ +# Test Report + +## 测试场景 +1. **基础发现与调用**: 验证服务启动后能自动注册到 Redis，并能通过 Caller 正确发起请求。 +2. **实时同步**: 验证通过 Redis PUBLISH 更新节点信息后，客户端能实时感知并更新本地节点列表。 +3. **故障剔除**: 验证当节点调用持续失败时，能自动从本地列表中剔除。 +4. **环境变量配置**: 验证 `EasyStart` 结合环境变量的启动流程。 + +## 测试结果 +- **Unit Tests**: `go test -v ./...` + - `TestDiscover`: PASS + - `TestEasyStart`: PASS + +## Benchmark +- 待补充（Discover 主要性能开销在负载均衡算法选择，单次选择耗时极低）。 diff --git a/go.mod b/go.mod new file mode 100644 index 0000000..09ab995 --- /dev/null +++ b/go.mod @@ -0,0 +1,29 @@ +module apigo.cc/go/discover + +go 1.25.0 + +require ( + apigo.cc/go/cast v1.2.6 + apigo.cc/go/config v1.0.4 + apigo.cc/go/http v1.0.3 + apigo.cc/go/id v1.0.4 + apigo.cc/go/log v1.0.2 + apigo.cc/go/redis v1.0.2 + github.com/gorilla/websocket v1.5.3 +) + +require ( + apigo.cc/go/convert v1.0.4 // indirect + apigo.cc/go/crypto v1.0.4 // indirect + apigo.cc/go/encoding v1.0.4 // indirect + apigo.cc/go/file v1.0.4 // indirect + apigo.cc/go/rand v1.0.4 // indirect + apigo.cc/go/safe v1.0.4 // indirect + apigo.cc/go/shell v1.0.4 // indirect + github.com/gomodule/redigo v1.9.3 // indirect + golang.org/x/crypto v0.50.0 // indirect + golang.org/x/net v0.53.0 // indirect + golang.org/x/sys v0.43.0 // indirect + golang.org/x/text v0.36.0 // indirect + gopkg.in/yaml.v3 v3.0.1 // indirect +) diff --git a/go.sum b/go.sum new file mode 100644 index 0000000..5389192 --- /dev/null +++ b/go.sum @@ -0,0 +1,48 @@ +apigo.cc/go/cast v1.2.6 h1:xnWiaQAGsRCrnu1p8fIFQfg5HFSc7CxR+3ItiDIDMaY= +apigo.cc/go/cast v1.2.6/go.mod h1:lGlwImiOvHxG7buyMWhFzcdvQzmSaoKbmr7bcDfUpHk= +apigo.cc/go/config v1.0.4 h1:WG9zrQkqfFPkrKIL7RNvvAbbkuUBt1Av11ZP/aIfldM= +apigo.cc/go/config v1.0.4/go.mod h1:obryzJiK6j7lQex/58d5eWYOGx5O5IABguqNWxyyXJo= +apigo.cc/go/convert v1.0.4 h1:5+qPjC3dlPB59GnWZRlmthxcaXQtKvN+iOuiLdJ1GvQ= +apigo.cc/go/convert v1.0.4/go.mod h1:Hp+geeSyhqg/zwIKPOrDoceIREzcwM14t1I5q/dtbfU= +apigo.cc/go/crypto v1.0.4 h1:VPUyHCH2N3LLEgdpwUc+DQssNHzLlxVzLNRa0Jm6O4o= +apigo.cc/go/crypto v1.0.4/go.mod h1:5sI8BLw6YHZfDReYwCO3TFD2LKm36HMdLg1S5oPv/QU= +apigo.cc/go/encoding v1.0.4 h1:aezB0J/qFuHs6iXkbtuJP5JIHUtmjsr5SFb0NNvbObY= +apigo.cc/go/encoding v1.0.4/go.mod h1:V5CgT7rBbCxy+uCU20q0ptcNNRSgMtpA8cNOs6r8IeI= +apigo.cc/go/file v1.0.4 h1:qCKegV7OYh7r0qc3jZjGA/aKh0vIHgmr1OEbhfEmGX8= +apigo.cc/go/file v1.0.4/go.mod h1:C9gNo7386iA21OiBmuWh6CznKWlVBDFkhE4f0H0Susg= +apigo.cc/go/http v1.0.3 h1:c19ppdb7gR9aIPeY3qOjOj4X3+jZLXln76jTTj7i4vM= +apigo.cc/go/http v1.0.3/go.mod h1:oHQYlBLN6u53C2t1BihxT7cnUQd+zLTAYr3ALjWUkpg= +apigo.cc/go/id v1.0.4 h1:w+JSdeVit52iefIUolrh1qLEZS9XqHNKr1UygFcgv+s= +apigo.cc/go/id v1.0.4/go.mod h1:kg7QuceAKtGNzGWt0+pIIh8Qom1eMSWGb8+0Yhi/QVY= +apigo.cc/go/log v1.0.2 h1:OY6T3SC28blDNkMpdRvDK2N4sGdriAB9DBItGl/qOos= +apigo.cc/go/log v1.0.2/go.mod h1:tvPgFpebY9Wf/DlqMHZ0ZjxDp9AaQTywOQKvtBaNqNo= +apigo.cc/go/rand v1.0.4 h1:we070eWSL0dB8NEMaWjXj43+EekXQTm/h0kKpZ/frqw= +apigo.cc/go/rand v1.0.4/go.mod h1:mZ/4Soa3bk+XvDaqPWJuUe1bfEi4eThBj1XmEAuYxsk= +apigo.cc/go/redis v1.0.2 h1:gWBrL/6eDxtouTFSZrPKQNdEg1AZr2aKTpCOhwim3dI= +apigo.cc/go/redis v1.0.2/go.mod h1:auQ3cyORgD67HF5dNvZ1lA8bqMH1xIbnuKBuZWclNy4= +apigo.cc/go/safe v1.0.4 h1:07pRSdEHprF/2v6SsqAjICYFoeLcqjjvHGEdh6Dzrzg= +apigo.cc/go/safe v1.0.4/go.mod h1:o568sHS5rTRSVPmhxWod0tGdc+8l1KjidsNY1/OVZr0= +apigo.cc/go/shell v1.0.4 h1:EL9zjI39YBe1h+kRYQeAi/8zVGHe5W198DYYN7cENiY= +apigo.cc/go/shell v1.0.4/go.mod h1:N2gDkgK4tJ9TadD60/+gAGuWxyVAWHs5YPBmytw6ELA= +github.com/davecgh/go-spew v1.1.1 h1:vj9j/u1bqnvCEfJOwUhtlOARqs3+rkHYY13jYWTU97c= +github.com/davecgh/go-spew v1.1.1/go.mod h1:J7Y8YcW2NihsgmVo/mv3lAwl/skON4iLHjSsI+c5H38= +github.com/gomodule/redigo v1.9.3 h1:dNPSXeXv6HCq2jdyWfjgmhBdqnR6PRO3m/G05nvpPC8= +github.com/gomodule/redigo v1.9.3/go.mod h1:KsU3hiK/Ay8U42qpaJk+kuNa3C+spxapWpM+ywhcgtw= +github.com/gorilla/websocket v1.5.3 h1:saDtZ6Pbx/0u+bgYQ3q96pZgCzfhKXGPqt7kZ72aNNg= +github.com/gorilla/websocket v1.5.3/go.mod h1:YR8l580nyteQvAITg2hZ9XVh4b55+EU/adAjf1fMHhE= +github.com/pmezard/go-difflib v1.0.0 h1:4DBwDE0NGyQoBHbLQYPwSUPoCMWR5BEzIk/f1lZbAQM= +github.com/pmezard/go-difflib v1.0.0/go.mod h1:iKH77koFhYxTK1pcRnkKkqfTogsbg7gZNVY4sRDYZ/4= +github.com/stretchr/testify v1.8.4 h1:CcVxjf3Q8PM0mHUKJCdn+eZZtm5yQwehR5yeSVQQcUk= +github.com/stretchr/testify v1.8.4/go.mod h1:sz/lmYIOXD/1dqDmKjjqLyZ2RngseejIcXlSw2iwfAo= +golang.org/x/crypto v0.50.0 h1:zO47/JPrL6vsNkINmLoo/PH1gcxpls50DNogFvB5ZGI= +golang.org/x/crypto v0.50.0/go.mod h1:3muZ7vA7PBCE6xgPX7nkzzjiUq87kRItoJQM1Yo8S+Q= +golang.org/x/net v0.53.0 h1:d+qAbo5L0orcWAr0a9JweQpjXF19LMXJE8Ey7hwOdUA= +golang.org/x/net v0.53.0/go.mod h1:JvMuJH7rrdiCfbeHoo3fCQU24Lf5JJwT9W3sJFulfgs= +golang.org/x/sys v0.43.0 h1:Rlag2XtaFTxp19wS8MXlJwTvoh8ArU6ezoyFsMyCTNI= +golang.org/x/sys v0.43.0/go.mod h1:4GL1E5IUh+htKOUEOaiffhrAeqysfVGipDYzABqnCmw= +golang.org/x/text v0.36.0 h1:JfKh3XmcRPqZPKevfXVpI1wXPTqbkE5f7JA92a55Yxg= +golang.org/x/text v0.36.0/go.mod h1:NIdBknypM8iqVmPiuco0Dh6P5Jcdk8lJL0CUebqK164= +gopkg.in/check.v1 v0.0.0-20161208181325-20d25e280405 h1:yhCVgyC4o1eVCa2tZl7eS0r+SDo693bJlVdllGtEeKM= +gopkg.in/check.v1 v0.0.0-20161208181325-20d25e280405/go.mod h1:Co6ibVJAznAaIkqp8huTwlJQCZ016jof/cbN4VW5Yz0= +gopkg.in/yaml.v3 v3.0.1 h1:fxVm/GzAzEWqLHuvctI91KS9hhNmmWOoWu0XTYJS7CA= +gopkg.in/yaml.v3 v3.0.1/go.mod h1:K4uyk7z7BCEPqu6E+C64Yfv1cQ7kz7rIZviUmN+EgEM=