blackteay/open-im-server
1
0
Fork 0
mirror of https://github.com/openimsdk/open-im-server.git synced 2025-10-27 05:52:29 +08:00
Code Issues Packages Projects Releases Wiki Activity

feat: compatible autoSetPorts (#2929)

Browse Source
This commit is contained in:
icey-yu 2024-12-06 17:25:42 +08:00 committed by OpenIM-Robot
parent 71dd84e88f
commit c07af4bb30
22 changed files with 51 additions and 260 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

1
config/openim-msggateway.yml
View File

@ -2,7 +2,6 @@ rpc:
# The IP address where this RPC service registers itself; if left blank, it defaults to the internal network IP
registerIP:
# autoSetPorts indicates whether to automatically set the ports
# if you use in kubernetes, set it to false
autoSetPorts: true
# List of ports that the RPC service listens on; configuring multiple ports will launch multiple instances. These must match the number of configured prometheus ports
# It will only take effect when autoSetPorts is set to false.

1
config/openim-push.yml
View File

@ -4,7 +4,6 @@ rpc:
# IP address that the RPC service listens on; setting to 0.0.0.0 listens on both internal and external IPs. If left blank, it automatically uses the internal network IP
listenIP: 0.0.0.0
# autoSetPorts indicates whether to automatically set the ports
# if you use in kubernetes, set it to false
autoSetPorts: true
# List of ports that the RPC service listens on; configuring multiple ports will launch multiple instances. These must match the number of configured prometheus ports
# It will only take effect when autoSetPorts is set to false.

1
config/openim-rpc-auth.yml
View File

@ -4,7 +4,6 @@ rpc:
# IP address that the RPC service listens on; setting to 0.0.0.0 listens on both internal and external IPs. If left blank, it automatically uses the internal network IP
listenIP: 0.0.0.0
# autoSetPorts indicates whether to automatically set the ports
# if you use in kubernetes, set it to false
autoSetPorts: true
# List of ports that the RPC service listens on; configuring multiple ports will launch multiple instances. These must match the number of configured prometheus ports
# It will only take effect when autoSetPorts is set to false.

1
config/openim-rpc-conversation.yml
View File

@ -4,7 +4,6 @@ rpc:
# IP address that the RPC service listens on; setting to 0.0.0.0 listens on both internal and external IPs. If left blank, it automatically uses the internal network IP
listenIP: 0.0.0.0
# autoSetPorts indicates whether to automatically set the ports
# if you use in kubernetes, set it to false
autoSetPorts: true
# List of ports that the RPC service listens on; configuring multiple ports will launch multiple instances. These must match the number of configured prometheus ports
# It will only take effect when autoSetPorts is set to false.

1
config/openim-rpc-friend.yml
View File

@ -4,7 +4,6 @@ rpc:
# IP address that the RPC service listens on; setting to 0.0.0.0 listens on both internal and external IPs. If left blank, it automatically uses the internal network IP
listenIP: 0.0.0.0
# autoSetPorts indicates whether to automatically set the ports
# if you use in kubernetes, set it to false
autoSetPorts: true
# List of ports that the RPC service listens on; configuring multiple ports will launch multiple instances. These must match the number of configured prometheus ports
# It will only take effect when autoSetPorts is set to false.

1
config/openim-rpc-group.yml
View File

@ -4,7 +4,6 @@ rpc:
# IP address that the RPC service listens on; setting to 0.0.0.0 listens on both internal and external IPs. If left blank, it automatically uses the internal network IP
listenIP: 0.0.0.0
# autoSetPorts indicates whether to automatically set the ports
# if you use in kubernetes, set it to false
autoSetPorts: true
# List of ports that the RPC service listens on; configuring multiple ports will launch multiple instances. These must match the number of configured prometheus ports
# It will only take effect when autoSetPorts is set to false.

1
config/openim-rpc-msg.yml
View File

@ -4,7 +4,6 @@ rpc:
# IP address that the RPC service listens on; setting to 0.0.0.0 listens on both internal and external IPs. If left blank, it automatically uses the internal network IP
listenIP: 0.0.0.0
# autoSetPorts indicates whether to automatically set the ports
# if you use in kubernetes, set it to false
autoSetPorts: true
# List of ports that the RPC service listens on; configuring multiple ports will launch multiple instances. These must match the number of configured prometheus ports
# It will only take effect when autoSetPorts is set to false.

1
config/openim-rpc-third.yml
View File

@ -4,7 +4,6 @@ rpc:
# IP address that the RPC service listens on; setting to 0.0.0.0 listens on both internal and external IPs. If left blank, it automatically uses the internal network IP
listenIP: 0.0.0.0
# autoSetPorts indicates whether to automatically set the ports
# if you use in kubernetes, set it to false
autoSetPorts: true
# List of ports that the RPC service listens on; configuring multiple ports will launch multiple instances. These must match the number of configured prometheus ports
# It will only take effect when autoSetPorts is set to false.

1
config/openim-rpc-user.yml
View File

@ -4,7 +4,6 @@ rpc:
# Listening IP; 0.0.0.0 means both internal and external IPs are listened to, if blank, the internal network IP is automatically obtained by default
listenIP: 0.0.0.0
# autoSetPorts indicates whether to automatically set the ports
# if you use in kubernetes, set it to false
autoSetPorts: true
# List of ports that the RPC service listens on; configuring multiple ports will launch multiple instances. These must match the number of configured prometheus ports
# It will only take effect when autoSetPorts is set to false.

3
internal/msggateway/hub_server.go
View File

@ -46,7 +46,8 @@ func (s *Server) InitServer(ctx context.Context, config *Config, disCov discover
func (s *Server) Start(ctx context.Context, index int, conf *Config) error {
return startrpc.Start(ctx, &conf.Discovery, &conf.MsgGateway.Prometheus, conf.MsgGateway.ListenIP,
index,
conf.MsgGateway.RPC.RegisterIP,
conf.MsgGateway.RPC.AutoSetPorts, conf.MsgGateway.RPC.Ports, index,
conf.Share.RpcRegisterName.MessageGateway,
&conf.Share,
conf,

4
internal/msggateway/init.go
View File

@ -39,9 +39,7 @@ type Config struct {
// Start run ws server.
func Start(ctx context.Context, index int, conf *Config) error {
conf.RuntimeEnv = runtimeenv.PrintRuntimeEnvironment()
log.CInfo(ctx, "MSG-GATEWAY server is initializing", "runtimeEnv", conf.RuntimeEnv, "autoSetPorts", conf.MsgGateway.RPC.AutoSetPorts,
log.CInfo(ctx, "MSG-GATEWAY server is initializing", "autoSetPorts", conf.MsgGateway.RPC.AutoSetPorts,
"rpcPorts", conf.MsgGateway.RPC.Ports,
"wsPort", conf.MsgGateway.LongConnSvr.Ports, "prometheusPorts", conf.MsgGateway.Prometheus.Ports)
wsPort, err := datautil.GetElemByIndex(conf.MsgGateway.LongConnSvr.Ports, index)

165
internal/tools/addr/addr.go
View File

@ -1,165 +0,0 @@
// addr provides functions to retrieve local IP addresses from device interfaces.
package addr
import (
"net"
"github.com/pkg/errors"
)
var (
// ErrIPNotFound no IP address found, and explicit IP not provided.
ErrIPNotFound = errors.New("no IP address found, and explicit IP not provided")
)
// IsLocal checks whether an IP belongs to one of the device's interfaces.
func IsLocal(addr string) bool {
// Extract the host
host, _, err := net.SplitHostPort(addr)
if err == nil {
addr = host
}
if addr == "localhost" {
return true
}
// Check against all local ips
for _, ip := range IPs() {
if addr == ip {
return true
}
}
return false
}
// Extract returns a valid IP address. If the address provided is a valid
// address, it will be returned directly. Otherwise, the available interfaces
// will be iterated over to find an IP address, preferably private.
func Extract(addr string) (string, error) {
// if addr is already specified then it's directly returned
if len(addr) > 0 && (addr != "0.0.0.0" && addr != "[::]" && addr != "::") {
return addr, nil
}
var (
addrs []net.Addr
loAddrs []net.Addr
)
ifaces, err := net.Interfaces()
if err != nil {
return "", errors.Wrap(err, "failed to get interfaces")
}
for _, iface := range ifaces {
ifaceAddrs, err := iface.Addrs()
if err != nil {
// ignore error, interface can disappear from system
continue
}
if iface.Flags&net.FlagLoopback != 0 {
loAddrs = append(loAddrs, ifaceAddrs...)
continue
}
addrs = append(addrs, ifaceAddrs...)
}
// Add loopback addresses to the end of the list
addrs = append(addrs, loAddrs...)
// Try to find private IP in list, public IP otherwise
ip, err := findIP(addrs)
if err != nil {
return "", err
}
return ip.String(), nil
}
// IPs returns all available interface IP addresses.
func IPs() []string {
ifaces, err := net.Interfaces()
if err != nil {
return nil
}
var ipAddrs []string
for _, i := range ifaces {
addrs, err := i.Addrs()
if err != nil {
continue
}
for _, addr := range addrs {
var ip net.IP
switch v := addr.(type) {
case *net.IPNet:
ip = v.IP
case *net.IPAddr:
ip = v.IP
}
if ip == nil {
continue
}
ipAddrs = append(ipAddrs, ip.String())
}
}
return ipAddrs
}
// findIP will return the first private IP available in the list.
// If no private IP is available it will return the first public IP, if present.
// If no public IP is available, it will return the first loopback IP, if present.
func findIP(addresses []net.Addr) (net.IP, error) {
var publicIP net.IP
var localIP net.IP
for _, rawAddr := range addresses {
var ip net.IP
switch addr := rawAddr.(type) {
case *net.IPAddr:
ip = addr.IP
case *net.IPNet:
ip = addr.IP
default:
continue
}
if ip.IsLoopback() {
if localIP == nil {
localIP = ip
}
continue
}
if !ip.IsPrivate() {
if publicIP == nil {
publicIP = ip
}
continue
}
// Return private IP if available
return ip, nil
}
// Return public or virtual IP
if len(publicIP) > 0 {
return publicIP, nil
}
// Return local IP
if len(localIP) > 0 {
return localIP, nil
}
return nil, ErrIPNotFound
}

1
pkg/common/cmd/auth.go
View File

@ -55,5 +55,6 @@ func (a *AuthRpcCmd) Exec() error {
func (a *AuthRpcCmd) runE() error {
return startrpc.Start(a.ctx, &a.authConfig.Discovery, &a.authConfig.RpcConfig.Prometheus, a.authConfig.RpcConfig.RPC.ListenIP,
a.authConfig.RpcConfig.RPC.RegisterIP, a.authConfig.RpcConfig.RPC.AutoSetPorts, a.authConfig.RpcConfig.RPC.Ports,
a.Index(), a.authConfig.Share.RpcRegisterName.Auth, &a.authConfig.Share, a.authConfig, auth.Start)
}

1
pkg/common/cmd/conversation.go
View File

@ -57,5 +57,6 @@ func (a *ConversationRpcCmd) Exec() error {
func (a *ConversationRpcCmd) runE() error {
return startrpc.Start(a.ctx, &a.conversationConfig.Discovery, &a.conversationConfig.RpcConfig.Prometheus, a.conversationConfig.RpcConfig.RPC.ListenIP,
a.conversationConfig.RpcConfig.RPC.RegisterIP, a.conversationConfig.RpcConfig.RPC.AutoSetPorts, a.conversationConfig.RpcConfig.RPC.Ports,
a.Index(), a.conversationConfig.Share.RpcRegisterName.Conversation, &a.conversationConfig.Share, a.conversationConfig, conversation.Start)
}

1
pkg/common/cmd/friend.go
View File

@ -58,5 +58,6 @@ func (a *FriendRpcCmd) Exec() error {
func (a *FriendRpcCmd) runE() error {
return startrpc.Start(a.ctx, &a.relationConfig.Discovery, &a.relationConfig.RpcConfig.Prometheus, a.relationConfig.RpcConfig.RPC.ListenIP,
a.relationConfig.RpcConfig.RPC.RegisterIP, a.relationConfig.RpcConfig.RPC.AutoSetPorts, a.relationConfig.RpcConfig.RPC.Ports,
a.Index(), a.relationConfig.Share.RpcRegisterName.Friend, &a.relationConfig.Share, a.relationConfig, relation.Start)
}

1
pkg/common/cmd/group.go
View File

@ -59,5 +59,6 @@ func (a *GroupRpcCmd) Exec() error {
func (a *GroupRpcCmd) runE() error {
return startrpc.Start(a.ctx, &a.groupConfig.Discovery, &a.groupConfig.RpcConfig.Prometheus, a.groupConfig.RpcConfig.RPC.ListenIP,
a.groupConfig.RpcConfig.RPC.RegisterIP, a.groupConfig.RpcConfig.RPC.AutoSetPorts, a.groupConfig.RpcConfig.RPC.Ports,
a.Index(), a.groupConfig.Share.RpcRegisterName.Group, &a.groupConfig.Share, a.groupConfig, group.Start, versionctx.EnableVersionCtx())
}

1
pkg/common/cmd/msg.go
View File

@ -59,5 +59,6 @@ func (a *MsgRpcCmd) Exec() error {
func (a *MsgRpcCmd) runE() error {
return startrpc.Start(a.ctx, &a.msgConfig.Discovery, &a.msgConfig.RpcConfig.Prometheus, a.msgConfig.RpcConfig.RPC.ListenIP,
a.msgConfig.RpcConfig.RPC.RegisterIP, a.msgConfig.RpcConfig.RPC.AutoSetPorts, a.msgConfig.RpcConfig.RPC.Ports,
a.Index(), a.msgConfig.Share.RpcRegisterName.Msg, &a.msgConfig.Share, a.msgConfig, msg.Start)
}

1
pkg/common/cmd/push.go
View File

@ -59,5 +59,6 @@ func (a *PushRpcCmd) Exec() error {
func (a *PushRpcCmd) runE() error {
return startrpc.Start(a.ctx, &a.pushConfig.Discovery, &a.pushConfig.RpcConfig.Prometheus, a.pushConfig.RpcConfig.RPC.ListenIP,
a.pushConfig.RpcConfig.RPC.RegisterIP, a.pushConfig.RpcConfig.RPC.AutoSetPorts, a.pushConfig.RpcConfig.RPC.Ports,
a.Index(), a.pushConfig.Share.RpcRegisterName.Push, &a.pushConfig.Share, a.pushConfig, push.Start)
}

1
pkg/common/cmd/third.go
View File

@ -58,5 +58,6 @@ func (a *ThirdRpcCmd) Exec() error {
func (a *ThirdRpcCmd) runE() error {
return startrpc.Start(a.ctx, &a.thirdConfig.Discovery, &a.thirdConfig.RpcConfig.Prometheus, a.thirdConfig.RpcConfig.RPC.ListenIP,
a.thirdConfig.RpcConfig.RPC.RegisterIP, a.thirdConfig.RpcConfig.RPC.AutoSetPorts, a.thirdConfig.RpcConfig.RPC.Ports,
a.Index(), a.thirdConfig.Share.RpcRegisterName.Third, &a.thirdConfig.Share, a.thirdConfig, third.Start)
}

1
pkg/common/cmd/user.go
View File

@ -59,5 +59,6 @@ func (a *UserRpcCmd) Exec() error {
func (a *UserRpcCmd) runE() error {
return startrpc.Start(a.ctx, &a.userConfig.Discovery, &a.userConfig.RpcConfig.Prometheus, a.userConfig.RpcConfig.RPC.ListenIP,
a.userConfig.RpcConfig.RPC.RegisterIP, a.userConfig.RpcConfig.RPC.AutoSetPorts, a.userConfig.RpcConfig.RPC.Ports,
a.Index(), a.userConfig.Share.RpcRegisterName.User, &a.userConfig.Share, a.userConfig, user.Start)
}

10
pkg/common/config/config.go
View File

@ -179,6 +179,7 @@ type Prometheus struct {
type MsgGateway struct {
RPC struct {
RegisterIP string `mapstructure:"registerIP"`
AutoSetPorts bool `mapstructure:"autoSetPorts"`
Ports []int `mapstructure:"ports"`
} `mapstructure:"rpc"`
Prometheus Prometheus `mapstructure:"prometheus"`
@ -199,6 +200,8 @@ type Push struct {
RPC struct {
RegisterIP string `mapstructure:"registerIP"`
ListenIP string `mapstructure:"listenIP"`
AutoSetPorts bool `mapstructure:"autoSetPorts"`
Ports []int `mapstructure:"ports"`
} `mapstructure:"rpc"`
Prometheus Prometheus `mapstructure:"prometheus"`
MaxConcurrentWorkers int `mapstructure:"maxConcurrentWorkers"`
@ -233,6 +236,7 @@ type Auth struct {
RPC struct {
RegisterIP string `mapstructure:"registerIP"`
ListenIP string `mapstructure:"listenIP"`
AutoSetPorts bool `mapstructure:"autoSetPorts"`
Ports []int `mapstructure:"ports"`
} `mapstructure:"rpc"`
Prometheus Prometheus `mapstructure:"prometheus"`
@ -245,6 +249,7 @@ type Conversation struct {
RPC struct {
RegisterIP string `mapstructure:"registerIP"`
ListenIP string `mapstructure:"listenIP"`
AutoSetPorts bool `mapstructure:"autoSetPorts"`
Ports []int `mapstructure:"ports"`
} `mapstructure:"rpc"`
Prometheus Prometheus `mapstructure:"prometheus"`
@ -254,6 +259,7 @@ type Friend struct {
RPC struct {
RegisterIP string `mapstructure:"registerIP"`
ListenIP string `mapstructure:"listenIP"`
AutoSetPorts bool `mapstructure:"autoSetPorts"`
Ports []int `mapstructure:"ports"`
} `mapstructure:"rpc"`
Prometheus Prometheus `mapstructure:"prometheus"`
@ -263,6 +269,7 @@ type Group struct {
RPC struct {
RegisterIP string `mapstructure:"registerIP"`
ListenIP string `mapstructure:"listenIP"`
AutoSetPorts bool `mapstructure:"autoSetPorts"`
Ports []int `mapstructure:"ports"`
} `mapstructure:"rpc"`
Prometheus Prometheus `mapstructure:"prometheus"`
@ -273,6 +280,7 @@ type Msg struct {
RPC struct {
RegisterIP string `mapstructure:"registerIP"`
ListenIP string `mapstructure:"listenIP"`
AutoSetPorts bool `mapstructure:"autoSetPorts"`
Ports []int `mapstructure:"ports"`
} `mapstructure:"rpc"`
Prometheus Prometheus `mapstructure:"prometheus"`
@ -283,6 +291,7 @@ type Third struct {
RPC struct {
RegisterIP string `mapstructure:"registerIP"`
ListenIP string `mapstructure:"listenIP"`
AutoSetPorts bool `mapstructure:"autoSetPorts"`
Ports []int `mapstructure:"ports"`
} `mapstructure:"rpc"`
Prometheus Prometheus `mapstructure:"prometheus"`
@ -332,6 +341,7 @@ type User struct {
RPC struct {
RegisterIP string `mapstructure:"registerIP"`
ListenIP string `mapstructure:"listenIP"`
AutoSetPorts bool `mapstructure:"autoSetPorts"`
Ports []int `mapstructure:"ports"`
} `mapstructure:"rpc"`
Prometheus Prometheus `mapstructure:"prometheus"`

62
pkg/common/startrpc/start.go
View File

@ -16,6 +16,7 @@ package startrpc
import (
"context"
"errors"
"fmt"
"net"
"net/http"
@ -24,13 +25,6 @@ import (
"syscall"
"time"
"github.com/openimsdk/open-im-server/v3/pkg/common/config"
"github.com/openimsdk/tools/utils/datautil"
"github.com/openimsdk/tools/utils/runtimeenv"
"google.golang.org/grpc/status"
"strconv"
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"
@ -53,8 +47,6 @@ func Start[T any](ctx context.Context, discovery *config.Discovery, prometheusCo
netErr error
)
runTimeEnv := runtimeenv.PrintRuntimeEnvironment()
if !autoSetPorts {
rpcPort, err := datautil.GetElemByIndex(rpcPorts, index)
if err != nil {
@ -81,7 +73,7 @@ func Start[T any](ctx context.Context, discovery *config.Discovery, prometheusCo
if err != nil {
return err
}
cs := prommetrics.GetGrpcCusMetrics(rpcRegisterName, discovery)
cs := prommetrics.GetGrpcCusMetrics(rpcRegisterName, share)
go func() {
if err := prommetrics.RpcInit(cs, prometheusPort); err != nil && !errors.Is(err, http.ErrServerClosed) {
netErr = errs.WrapMsg(err, fmt.Sprintf("rpc %s prometheus start err: %d", rpcRegisterName, prometheusPort))
@ -99,7 +91,6 @@ func Start[T any](ctx context.Context, discovery *config.Discovery, prometheusCo
options = append(options, mw.GrpcServer())
}
rpcTcpAddr := net.JoinHostPort(network.GetListenIP(listenIP), "0")
listener, err := net.Listen(
"tcp",
rpcTcpAddr,
@ -108,8 +99,8 @@ func Start[T any](ctx context.Context, discovery *config.Discovery, prometheusCo
return errs.WrapMsg(err, "listen err", "rpcTcpAddr", rpcTcpAddr)
}
h, portStr, _ := net.SplitHostPort(listener.Addr().String())
host, _ := addr.Extract(h)
_, portStr, _ := net.SplitHostPort(listener.Addr().String())
registerIP = network.GetListenIP(registerIP)
port, _ := strconv.Atoi(portStr)
log.CInfo(ctx, "RPC server is initializing", "runTimeEnv", runTimeEnv, "rpcRegisterName", rpcRegisterName, "rpcPort", portStr,
@ -124,22 +115,6 @@ func Start[T any](ctx context.Context, discovery *config.Discovery, prometheusCo
defer client.Close()
client.AddOption(mw.GrpcClient(), grpc.WithTransportCredentials(insecure.NewCredentials()), grpc.WithDefaultServiceConfig(fmt.Sprintf(`{"LoadBalancingPolicy": "%s"}`, "round_robin")))
// var reg *prometheus.Registry
// var metric *grpcprometheus.ServerMetrics
if prometheusConfig.Enable {
// cusMetrics := prommetrics.GetGrpcCusMetrics(rpcRegisterName, share)
// reg, metric, _ = prommetrics.NewGrpcPromObj(cusMetrics)
// options = append(options, mw.GrpcServer(), grpc.StreamInterceptor(metric.StreamServerInterceptor()),
// grpc.UnaryInterceptor(metric.UnaryServerInterceptor()))
options = append(
options, mw.GrpcServer(),
prommetricsUnaryInterceptor(rpcRegisterName),
prommetricsStreamInterceptor(rpcRegisterName),
)
} else {
options = append(options, mw.GrpcServer())
}
srv := grpc.NewServer(options...)
err = rpcFn(ctx, config, client, srv)
@ -149,7 +124,7 @@ func Start[T any](ctx context.Context, discovery *config.Discovery, prometheusCo
err = client.Register(
rpcRegisterName,
host,
registerIP,
port,
grpc.WithTransportCredentials(insecure.NewCredentials()),
)
@ -157,33 +132,6 @@ func Start[T any](ctx context.Context, discovery *config.Discovery, prometheusCo
return err
}
var (
netDone = make(chan struct{}, 2)
netErr error
)
if prometheusConfig.Enable {
go func() {
prometheusPort, err := datautil.GetElemByIndex(prometheusConfig.Ports, index)
if err != nil {
netErr = err
netDone <- struct{}{}
return
}
cs := prommetrics.GetGrpcCusMetrics(rpcRegisterName, share)
if err := prommetrics.RpcInit(cs, prometheusPort); err != nil && err != http.ErrServerClosed {
netErr = errs.WrapMsg(err, fmt.Sprintf("rpc %s prometheus start err: %d", rpcRegisterName, prometheusPort))
netDone <- struct{}{}
}
// metric.InitializeMetrics(srv)
// Create a HTTP server for prometheus.
// httpServer = &http.Server{Handler: promhttp.HandlerFor(reg, promhttp.HandlerOpts{}), Addr: fmt.Sprintf("0.0.0.0:%d", prometheusPort)}
// if err := httpServer.ListenAndServe(); err != nil && err != http.ErrServerClosed {
// netErr = errs.WrapMsg(err, "prometheus start err", httpServer.Addr)
// netDone <- struct{}{}
// }
}()
}
go func() {
err := srv.Serve(listener)
if err != nil {