fix: add autoPort && prometheus port discovery

2026-05-10 20:15:59 +08:00 · 2024-12-10 11:43:19 +08:00
parent 7912ac0623
commit 1e68f99d11
38 changed files with 670 additions and 253 deletions
@@ -2,6 +2,7 @@ package api

 import (
 	"context"
+	"errors"
 	"fmt"
 	"net"
 	"net/http"
@@ -12,15 +13,20 @@ import (
 	"time"

 	"github.com/openimsdk/open-im-server/v3/pkg/common/config"
+	"github.com/openimsdk/tools/mw"
 	"github.com/openimsdk/tools/utils/datautil"
 	"github.com/openimsdk/tools/utils/network"
+	"google.golang.org/grpc"
+	"google.golang.org/grpc/credentials/insecure"

 	kdisc "github.com/openimsdk/open-im-server/v3/pkg/common/discoveryregister"
 	"github.com/openimsdk/open-im-server/v3/pkg/common/prommetrics"
 	"github.com/openimsdk/tools/discovery"
+	"github.com/openimsdk/tools/discovery/etcd"
 	"github.com/openimsdk/tools/errs"
 	"github.com/openimsdk/tools/log"
 	"github.com/openimsdk/tools/system/program"
+	"github.com/openimsdk/tools/utils/jsonutil"
 )

 type Config struct {
@@ -29,8 +35,8 @@ type Config struct {
 	Discovery config.Discovery
 }

-func Start(ctx context.Context, index int, config *Config) error {
-	apiPort, err := datautil.GetElemByIndex(config.API.Api.Ports, index)
+func Start(ctx context.Context, index int, cfg *Config) error {
+	apiPort, err := datautil.GetElemByIndex(cfg.API.Api.Ports, index)
 	if err != nil {
 		return err
 	}
@@ -38,12 +44,14 @@ func Start(ctx context.Context, index int, config *Config) error {
 	var client discovery.SvcDiscoveryRegistry

 	// Determine whether zk is passed according to whether it is a clustered deployment
-	client, err = kdisc.NewDiscoveryRegister(&config.Discovery, &config.Share, []string{
-		config.Share.RpcRegisterName.MessageGateway,
+	client, err = kdisc.NewDiscoveryRegister(&cfg.Discovery, &cfg.Share, []string{
+		cfg.Share.RpcRegisterName.MessageGateway,
 	})
 	if err != nil {
 		return errs.WrapMsg(err, "failed to register discovery service")
 	}
+	client.AddOption(mw.GrpcClient(), grpc.WithTransportCredentials(insecure.NewCredentials()),
+		grpc.WithDefaultServiceConfig(fmt.Sprintf(`{"LoadBalancingPolicy": "%s"}`, "round_robin")))

 	var (
 		netDone        = make(chan struct{}, 1)
@@ -51,32 +59,73 @@ func Start(ctx context.Context, index int, config *Config) error {
 		prometheusPort int
 	)

-	router, err := newGinRouter(ctx, client, config)
+	router, err := newGinRouter(ctx, client, cfg)
 	if err != nil {
 		return err
 	}
-	if config.API.Prometheus.Enable {
-		go func() {
-			prometheusPort, err = datautil.GetElemByIndex(config.API.Prometheus.Ports, index)
+	registerIP, err := network.GetRpcRegisterIP("")
+	if err != nil {
+		return err
+	}
+
+	getAutoPort := func() (net.Listener, int, error) {
+		registerAddr := net.JoinHostPort(registerIP, "0")
+		listener, err := net.Listen("tcp", registerAddr)
+		if err != nil {
+			return nil, 0, errs.WrapMsg(err, "listen err", "registerAddr", registerAddr)
+		}
+		_, portStr, _ := net.SplitHostPort(listener.Addr().String())
+		port, _ := strconv.Atoi(portStr)
+		return listener, port, nil
+	}
+
+	if cfg.API.Prometheus.AutoSetPorts && cfg.Discovery.Enable != config.ETCD {
+		return errs.New("only etcd support autoSetPorts", "RegisterName", "api").Wrap()
+	}
+
+	if cfg.API.Prometheus.Enable {
+		var (
+			listener net.Listener
+		)
+
+		if cfg.API.Prometheus.AutoSetPorts {
+			listener, prometheusPort, err = getAutoPort()
 			if err != nil {
-				netErr = err
-				netDone <- struct{}{}
-				return
+				return err
 			}
-			if err := prommetrics.ApiInit(prometheusPort); err != nil && err != http.ErrServerClosed {
+
+			etcdClient := client.(*etcd.SvcDiscoveryRegistryImpl).GetClient()
+
+			_, err = etcdClient.Put(ctx, prommetrics.BuildDiscoveryKey(prommetrics.APIKeyName), jsonutil.StructToJsonString(prommetrics.BuildDefaultTarget(registerIP, prometheusPort)))
+			if err != nil {
+				return errs.WrapMsg(err, "etcd put err")
+			}
+		} else {
+			prometheusPort, err = datautil.GetElemByIndex(cfg.API.Prometheus.Ports, index)
+			if err != nil {
+				return err
+			}
+			listener, err = net.Listen("tcp", fmt.Sprintf(":%d", prometheusPort))
+			if err != nil {
+				return errs.WrapMsg(err, "listen err", "addr", fmt.Sprintf(":%d", prometheusPort))
+			}
+		}
+
+		go func() {
+			if err := prommetrics.ApiInit(listener); err != nil && !errors.Is(err, http.ErrServerClosed) {
 				netErr = errs.WrapMsg(err, fmt.Sprintf("api prometheus start err: %d", prometheusPort))
 				netDone <- struct{}{}
 			}
 		}()

 	}
-	address := net.JoinHostPort(network.GetListenIP(config.API.Api.ListenIP), strconv.Itoa(apiPort))
+	address := net.JoinHostPort(network.GetListenIP(cfg.API.Api.ListenIP), strconv.Itoa(apiPort))

 	server := http.Server{Addr: address, Handler: router}
 	log.CInfo(ctx, "API server is initializing", "address", address, "apiPort", apiPort, "prometheusPort", prometheusPort)
 	go func() {
 		err = server.ListenAndServe()
-		if err != nil && err != http.ErrServerClosed {
+		if err != nil && !errors.Is(err, http.ErrServerClosed) {
 			netErr = errs.WrapMsg(err, fmt.Sprintf("api start err: %s", server.Addr))
 			netDone <- struct{}{}

@@ -0,0 +1,114 @@
+package api
+
+import (
+	"encoding/json"
+	"net/http"
+
+	"github.com/gin-gonic/gin"
+	"github.com/openimsdk/open-im-server/v3/pkg/common/config"
+	"github.com/openimsdk/open-im-server/v3/pkg/common/prommetrics"
+	"github.com/openimsdk/tools/apiresp"
+	"github.com/openimsdk/tools/discovery"
+	"github.com/openimsdk/tools/discovery/etcd"
+	"github.com/openimsdk/tools/errs"
+	"github.com/openimsdk/tools/log"
+	clientv3 "go.etcd.io/etcd/client/v3"
+)
+
+type PrometheusDiscoveryApi struct {
+	config *Config
+	client *clientv3.Client
+}
+
+func NewPrometheusDiscoveryApi(cfg *Config, client discovery.SvcDiscoveryRegistry) *PrometheusDiscoveryApi {
+	api := &PrometheusDiscoveryApi{
+		config: cfg,
+	}
+	if cfg.Discovery.Enable == config.ETCD {
+		api.client = client.(*etcd.SvcDiscoveryRegistryImpl).GetClient()
+	}
+	return api
+}
+
+func (p *PrometheusDiscoveryApi) Enable(c *gin.Context) {
+	if p.config.Discovery.Enable != config.ETCD {
+		c.JSON(http.StatusOK, []struct{}{})
+		c.Abort()
+	}
+}
+
+func (p *PrometheusDiscoveryApi) discovery(c *gin.Context, key string) {
+	eResp, err := p.client.Get(c, prommetrics.BuildDiscoveryKey(key))
+	if err != nil {
+		// Log and respond with an error if preparation fails.
+		apiresp.GinError(c, errs.WrapMsg(err, "etcd get err"))
+		return
+	}
+	if len(eResp.Kvs) == 0 {
+		c.JSON(http.StatusOK, []*prommetrics.Target{})
+	}
+
+	var (
+		resp = &prommetrics.RespTarget{
+			Targets: make([]string, 0, len(eResp.Kvs)),
+		}
+	)
+
+	for i := range eResp.Kvs {
+		var target prommetrics.Target
+		err = json.Unmarshal(eResp.Kvs[i].Value, &target)
+		if err != nil {
+			log.ZError(c, "prometheus unmarshal err", errs.Wrap(err))
+		}
+		resp.Targets = append(resp.Targets, target.Target)
+		if resp.Labels == nil {
+			resp.Labels = target.Labels
+		}
+	}
+
+	c.JSON(200, []*prommetrics.RespTarget{resp})
+}
+
+func (p *PrometheusDiscoveryApi) Api(c *gin.Context) {
+	p.discovery(c, prommetrics.APIKeyName)
+}
+
+func (p *PrometheusDiscoveryApi) User(c *gin.Context) {
+	p.discovery(c, p.config.Share.RpcRegisterName.User)
+}
+
+func (p *PrometheusDiscoveryApi) Group(c *gin.Context) {
+	p.discovery(c, p.config.Share.RpcRegisterName.Group)
+}
+
+func (p *PrometheusDiscoveryApi) Msg(c *gin.Context) {
+	p.discovery(c, p.config.Share.RpcRegisterName.Msg)
+}
+
+func (p *PrometheusDiscoveryApi) Friend(c *gin.Context) {
+	p.discovery(c, p.config.Share.RpcRegisterName.Friend)
+}
+
+func (p *PrometheusDiscoveryApi) Conversation(c *gin.Context) {
+	p.discovery(c, p.config.Share.RpcRegisterName.Conversation)
+}
+
+func (p *PrometheusDiscoveryApi) Third(c *gin.Context) {
+	p.discovery(c, p.config.Share.RpcRegisterName.Third)
+}
+
+func (p *PrometheusDiscoveryApi) Auth(c *gin.Context) {
+	p.discovery(c, p.config.Share.RpcRegisterName.Auth)
+}
+
+func (p *PrometheusDiscoveryApi) Push(c *gin.Context) {
+	p.discovery(c, p.config.Share.RpcRegisterName.Push)
+}
+
+func (p *PrometheusDiscoveryApi) MessageGateway(c *gin.Context) {
+	p.discovery(c, p.config.Share.RpcRegisterName.MessageGateway)
+}
+
+func (p *PrometheusDiscoveryApi) MessageTransfer(c *gin.Context) {
+	p.discovery(c, prommetrics.MessageTransferKeyName)
+}
@@ -2,7 +2,9 @@ package api

 import (
 	"context"
-	"fmt"
+	"net/http"
+	"strings"
+
 	"github.com/openimsdk/open-im-server/v3/pkg/rpcli"
 	pbAuth "github.com/openimsdk/protocol/auth"
 	"github.com/openimsdk/protocol/conversation"
@@ -11,8 +13,6 @@ import (
 	"github.com/openimsdk/protocol/relation"
 	"github.com/openimsdk/protocol/third"
 	"github.com/openimsdk/protocol/user"
-	"net/http"
-	"strings"

 	"github.com/openimsdk/open-im-server/v3/internal/api/jssdk"

@@ -21,9 +21,6 @@ import (
 	"github.com/gin-gonic/gin"
 	"github.com/gin-gonic/gin/binding"
 	"github.com/go-playground/validator/v10"
-	"google.golang.org/grpc"
-	"google.golang.org/grpc/credentials/insecure"
-
 	"github.com/openimsdk/open-im-server/v3/pkg/common/prommetrics"
 	"github.com/openimsdk/open-im-server/v3/pkg/common/servererrs"
 	"github.com/openimsdk/protocol/constant"
@@ -56,8 +53,6 @@ func prommetricsGin() gin.HandlerFunc {
 }

 func newGinRouter(ctx context.Context, client discovery.SvcDiscoveryRegistry, config *Config) (*gin.Engine, error) {
-	client.AddOption(mw.GrpcClient(), grpc.WithTransportCredentials(insecure.NewCredentials()),
-		grpc.WithDefaultServiceConfig(fmt.Sprintf(`{"LoadBalancingPolicy": "%s"}`, "round_robin")))
 	authConn, err := client.GetConn(ctx, config.Share.RpcRegisterName.Auth)
 	if err != nil {
 		return nil, err
@@ -100,12 +95,11 @@ func newGinRouter(ctx context.Context, client discovery.SvcDiscoveryRegistry, co
 	case BestSpeed:
 		r.Use(gzip.Gzip(gzip.BestSpeed))
 	}
-	r.Use(prommetricsGin(), gin.RecoveryWithWriter(gin.DefaultErrorWriter, mw.GinPanicErr), mw.CorsHandler(),
-		mw.GinParseOperationID(), GinParseToken(rpcli.NewAuthClient(authConn)))
-
+	r.Use(prommetricsGin(), gin.RecoveryWithWriter(gin.DefaultErrorWriter, mw.GinPanicErr), mw.CorsHandler(), mw.GinParseOperationID(), GinParseToken(rpcli.NewAuthClient(authConn)))
 	u := NewUserApi(user.NewUserClient(userConn), client, config.Share.RpcRegisterName)
+	m := NewMessageApi(msg.NewMsgClient(msgConn), rpcli.NewUserClient(userConn), config.Share.IMAdminUserID)
+	userRouterGroup := r.Group("/user")
 	{
-		userRouterGroup := r.Group("/user")
 		userRouterGroup.POST("/user_register", u.UserRegister)
 		userRouterGroup.POST("/update_user_info", u.UpdateUserInfo)
 		userRouterGroup.POST("/update_user_info_ex", u.UpdateUserInfoEx)
@@ -228,7 +222,6 @@ func newGinRouter(ctx context.Context, client discovery.SvcDiscoveryRegistry, co
 		objectGroup.GET("/*name", t.ObjectRedirect)
 	}
 	// Message
-	m := NewMessageApi(msg.NewMsgClient(msgConn), rpcli.NewUserClient(userConn), config.Share.IMAdminUserID)
 	{
 		msgGroup := r.Group("/msg")
 		msgGroup.POST("/newest_seq", m.GetSeq)
@@ -284,7 +277,21 @@ func newGinRouter(ctx context.Context, client discovery.SvcDiscoveryRegistry, co
 		jssdk.POST("/get_conversations", j.GetConversations)
 		jssdk.POST("/get_active_conversations", j.GetActiveConversations)
 	}
-
+	{
+		pd := NewPrometheusDiscoveryApi(config, client)
+		proDiscoveryGroup := r.Group("/prometheus_discovery", pd.Enable)
+		proDiscoveryGroup.GET("/api", pd.Api)
+		proDiscoveryGroup.GET("/user", pd.User)
+		proDiscoveryGroup.GET("/group", pd.Group)
+		proDiscoveryGroup.GET("/msg", pd.Msg)
+		proDiscoveryGroup.GET("/friend", pd.Friend)
+		proDiscoveryGroup.GET("/conversation", pd.Conversation)
+		proDiscoveryGroup.GET("/third", pd.Third)
+		proDiscoveryGroup.GET("/auth", pd.Auth)
+		proDiscoveryGroup.GET("/push", pd.Push)
+		proDiscoveryGroup.GET("/msg_gateway", pd.MessageGateway)
+		proDiscoveryGroup.GET("/msg_transfer", pd.MessageTransfer)
+	}
 	return r, nil
 }