Merge pull request #2318 from skiffer-git/etcd_3.7_2295

etcd naming and discovery

.env

@@ -4,7 +4,7 @@ REDIS_IMAGE=redis:7.0.0
 ZOOKEEPER_IMAGE=bitnami/zookeeper:3.8
 KAFKA_IMAGE=bitnami/kafka:3.5.1
 MINIO_IMAGE=minio/minio:RELEASE.2024-01-11T07-46-16Z
-
+ETCD_IMAGE=quay.io/coreos/etcd:v3.5.13
 
 OPENIM_WEB_FRONT_IMAGE=openim/openim-web-front:release-v3.5.1
 OPENIM_ADMIN_FRONT_IMAGE=openim/openim-admin-front:release-v1.7

config/discovery.yml

@@ -0,0 +1,13 @@
+enable: "etcd"
+etcd:
+  rootDirectory: openim
+  address: [ localhost:12379 ]
+  username: ''
+  password: ''
+
+zookeeper:
+  schema: openim
+  address: [ localhost:12181 ]
+  username: ''
+  password: ''
+

config/share.yml

@@ -1,5 +1,4 @@
 secret: openIM123
-env: zookeeper
 rpcRegisterName:
   user: user
   friend: friend

docker-compose.yml

@@ -58,6 +58,26 @@ services:
     networks:
       - openim
 
+  etcd:
+    image: "${ETCD_IMAGE}"
+    container_name: etcd
+    ports:
+      - "12379:2379"
+      - "12380:2380"
+    environment:
+      - ETCD_NAME=s1
+      - ETCD_DATA_DIR=/etcd-data
+      - ETCD_LISTEN_CLIENT_URLS=http://0.0.0.0:2379
+      - ETCD_ADVERTISE_CLIENT_URLS=http://0.0.0.0:2379
+      - ETCD_LISTEN_PEER_URLS=http://0.0.0.0:2380
+      - ETCD_INITIAL_ADVERTISE_PEER_URLS=http://0.0.0.0:2380
+      - ETCD_INITIAL_CLUSTER=s1=http://0.0.0.0:2380
+      - ETCD_INITIAL_CLUSTER_TOKEN=tkn
+      - ETCD_INITIAL_CLUSTER_STATE=new
+    restart: always
+    networks:
+      - openim
+
   kafka:
     image: "${KAFKA_IMAGE}"
     container_name: kafka

go.mod

@@ -58,6 +58,8 @@ require (
 	github.com/cespare/xxhash/v2 v2.2.0 // indirect
 	github.com/chenzhuoyu/base64x v0.0.0-20221115062448-fe3a3abad311 // indirect
 	github.com/clbanning/mxj v1.8.4 // indirect
+	github.com/coreos/go-semver v0.3.0 // indirect
+	github.com/coreos/go-systemd/v22 v22.3.2 // indirect
 	github.com/davecgh/go-spew v1.1.2-0.20180830191138-d8f796af33cc // indirect
 	github.com/dgryski/go-rendezvous v0.0.0-20200823014737-9f7001d12a5f // indirect
 	github.com/dustin/go-humanize v1.0.1 // indirect
@@ -137,6 +139,9 @@ require (
 	github.com/xdg-go/stringprep v1.0.4 // indirect
 	github.com/youmark/pkcs8 v0.0.0-20181117223130-1be2e3e5546d // indirect
 	github.com/yusufpapurcu/wmi v1.2.4 // indirect
+	go.etcd.io/etcd/api/v3 v3.5.13 // indirect
+	go.etcd.io/etcd/client/pkg/v3 v3.5.13 // indirect
+	go.etcd.io/etcd/client/v3 v3.5.13 // indirect
 	go.opencensus.io v0.24.0 // indirect
 	go.opentelemetry.io/contrib/instrumentation/google.golang.org/grpc/otelgrpc v0.47.0 // indirect
 	go.opentelemetry.io/contrib/instrumentation/net/http/otelhttp v0.47.0 // indirect

go.sum

@@ -47,6 +47,10 @@ github.com/client9/misspell v0.3.4/go.mod h1:qj6jICC3Q7zFZvVWo7KLAzC3yx5G7kyvSDk
 github.com/cncf/udpa/go v0.0.0-20191209042840-269d4d468f6f/go.mod h1:M8M6+tZqaGXZJjfX53e64911xZQV5JYwmTeXPW+k8Sc=
 github.com/cncf/xds/go v0.0.0-20231128003011-0fa0005c9caa h1:jQCWAUqqlij9Pgj2i/PB79y4KOPYVyFYdROxgaCwdTQ=
 github.com/cncf/xds/go v0.0.0-20231128003011-0fa0005c9caa/go.mod h1:x/1Gn8zydmfq8dk6e9PdstVsDgu9RuyIIJqAaF//0IM=
+github.com/coreos/go-semver v0.3.0 h1:wkHLiw0WNATZnSG7epLsujiMCgPAc9xhjJ4tgnAxmfM=
+github.com/coreos/go-semver v0.3.0/go.mod h1:nnelYz7RCh+5ahJtPPxZlU+153eP4D4r3EedlOD2RNk=
+github.com/coreos/go-systemd/v22 v22.3.2 h1:D9/bQk5vlXQFZ6Kwuu6zaiXJ9oTPe68++AzAJc1DzSI=
+github.com/coreos/go-systemd/v22 v22.3.2/go.mod h1:Y58oyj3AT4RCenI/lSvhwexgC+NSVTIJ3seZv2GcEnc=
 github.com/cpuguy83/go-md2man/v2 v2.0.3/go.mod h1:tgQtvFlXSQOSOSIRvRPT7W67SCa46tRHOmNcaadrF8o=
 github.com/davecgh/go-spew v1.1.0/go.mod h1:J7Y8YcW2NihsgmVo/mv3lAwl/skON4iLHjSsI+c5H38=
 github.com/davecgh/go-spew v1.1.1/go.mod h1:J7Y8YcW2NihsgmVo/mv3lAwl/skON4iLHjSsI+c5H38=
@@ -111,6 +115,7 @@ github.com/go-zookeeper/zk v1.0.3 h1:7M2kwOsc//9VeeFiPtf+uSJlVpU66x9Ba5+8XK7/TDg
 github.com/go-zookeeper/zk v1.0.3/go.mod h1:nOB03cncLtlp4t+UAkGSV+9beXP/akpekBwL+UX1Qcw=
 github.com/goccy/go-json v0.10.2 h1:CrxCmQqYDkv1z7lO7Wbh2HN93uovUHgrECaO5ZrCXAU=
 github.com/goccy/go-json v0.10.2/go.mod h1:6MelG93GURQebXPDq3khkgXZkazVtN9CRI+MGFi0w8I=
+github.com/godbus/dbus/v5 v5.0.4/go.mod h1:xhWf0FNVPg57R7Z0UbKHbJfkEywrmjJnf7w5xrFpKfA=
 github.com/gogo/protobuf v1.3.2 h1:Ov1cvc58UF3b5XjBnZv7+opcTcQFZebYjWzi34vdm4Q=
 github.com/gogo/protobuf v1.3.2/go.mod h1:P1XiOD3dCwIKUDQYPy72D8LYyHL2YPYrpS2s69NZV8Q=
 github.com/golang-jwt/jwt/v4 v4.5.0 h1:7cYmW1XlMY7h7ii7UhUyChSgS5wUJEnm9uZVTGqOWzg=
@@ -378,6 +383,12 @@ github.com/yuin/goldmark v1.2.1/go.mod h1:3hX8gzYuyVAZsxl0MRgGTJEmQBFcNTphYh9dec
 github.com/yuin/goldmark v1.4.13/go.mod h1:6yULJ656Px+3vBD8DxQVa3kxgyrAnzto9xy5taEt/CY=
 github.com/yusufpapurcu/wmi v1.2.4 h1:zFUKzehAFReQwLys1b/iSMl+JQGSCSjtVqQn9bBrPo0=
 github.com/yusufpapurcu/wmi v1.2.4/go.mod h1:SBZ9tNy3G9/m5Oi98Zks0QjeHVDvuK0qfxQmPyzfmi0=
+go.etcd.io/etcd/api/v3 v3.5.13 h1:8WXU2/NBge6AUF1K1gOexB6e07NgsN1hXK0rSTtgSp4=
+go.etcd.io/etcd/api/v3 v3.5.13/go.mod h1:gBqlqkcMMZMVTMm4NDZloEVJzxQOQIls8splbqBDa0c=
+go.etcd.io/etcd/client/pkg/v3 v3.5.13 h1:RVZSAnWWWiI5IrYAXjQorajncORbS0zI48LQlE2kQWg=
+go.etcd.io/etcd/client/pkg/v3 v3.5.13/go.mod h1:XxHT4u1qU12E2+po+UVPrEeL94Um6zL58ppuJWXSAB8=
+go.etcd.io/etcd/client/v3 v3.5.13 h1:o0fHTNJLeO0MyVbc7I3fsCf6nrOqn5d+diSarKnB2js=
+go.etcd.io/etcd/client/v3 v3.5.13/go.mod h1:cqiAeY8b5DEEcpxvgWKsbLIWNM/8Wy2xJSDMtioMcoI=
 go.mongodb.org/mongo-driver v1.14.0 h1:P98w8egYRjYe3XDjxhYJagTokP/H6HzlsnojRgZRd80=
 go.mongodb.org/mongo-driver v1.14.0/go.mod h1:Vzb0Mk/pa7e6cWw85R4F/endUC3u0U9jGcNU603k65c=
 go.opencensus.io v0.24.0 h1:y73uSU6J157QMP2kn2r30vwW1A2W2WFwSCGnAVxeaD0=

internal/api/init.go

@@ -38,28 +38,25 @@ import (
 )
 
 type Config struct {
-	RpcConfig          config.API
-	MongodbConfig      config.Mongo
-	ZookeeperConfig    config.ZooKeeper
-	NotificationConfig config.Notification
-	Share              config.Share
-	MinioConfig        config.Minio
+	API       config.API
+	Share     config.Share
+	Discovery config.Discovery
 }
 
 func Start(ctx context.Context, index int, config *Config) error {
-	apiPort, err := datautil.GetElemByIndex(config.RpcConfig.Api.Ports, index)
+	apiPort, err := datautil.GetElemByIndex(config.API.Api.Ports, index)
 	if err != nil {
 		return err
 	}
-	prometheusPort, err := datautil.GetElemByIndex(config.RpcConfig.Prometheus.Ports, index)
+	prometheusPort, err := datautil.GetElemByIndex(config.API.Prometheus.Ports, index)
 	if err != nil {
 		return err
 	}
 
 	var client discovery.SvcDiscoveryRegistry
 
 	// Determine whether zk is passed according to whether it is a clustered deployment
-	client, err = kdisc.NewDiscoveryRegister(&config.ZookeeperConfig, &config.Share)
+	client, err = kdisc.NewDiscoveryRegister(&config.Discovery, &config.Share)
 	if err != nil {
 		return errs.WrapMsg(err, "failed to register discovery service")
 	}
@@ -70,7 +67,7 @@ func Start(ctx context.Context, index int, config *Config) error {
 	)
 
 	router := newGinRouter(client, config)
-	if config.RpcConfig.Prometheus.Enable {
+	if config.API.Prometheus.Enable {
 		go func() {
 			p := ginprom.NewPrometheus("app", prommetrics.GetGinCusMetrics("Api"))
 			p.SetListenAddress(fmt.Sprintf(":%d", prometheusPort))
@@ -81,7 +78,7 @@ func Start(ctx context.Context, index int, config *Config) error {
 		}()
 
 	}
-	address := net.JoinHostPort(network.GetListenIP(config.RpcConfig.Api.ListenIP), strconv.Itoa(apiPort))
+	address := net.JoinHostPort(network.GetListenIP(config.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)

internal/api/router.go

@@ -34,7 +34,7 @@ func newGinRouter(disCov discovery.SvcDiscoveryRegistry, config *Config) *gin.En
 	messageRpc := rpcclient.NewMessage(disCov, config.Share.RpcRegisterName.Msg)
 	conversationRpc := rpcclient.NewConversation(disCov, config.Share.RpcRegisterName.Conversation)
 	authRpc := rpcclient.NewAuth(disCov, config.Share.RpcRegisterName.Auth)
-	thirdRpc := rpcclient.NewThird(disCov, config.Share.RpcRegisterName.Third, config.RpcConfig.Prometheus.GrafanaURL)
+	thirdRpc := rpcclient.NewThird(disCov, config.Share.RpcRegisterName.Third, config.API.Prometheus.GrafanaURL)
 
 	u := NewUserApi(*userRpc)
 	m := NewMessageApi(messageRpc, userRpc, config.Share.IMAdminUserID)

internal/msggateway/client.go

@@ -286,6 +286,7 @@ func (c *Client) KickOnlineMessage() error {
 	resp := Resp{
 		ReqIdentifier: WSKickOnlineMsg,
 	}
+	log.ZDebug(c.ctx, "KickOnlineMessage debug ")
 	err := c.writeBinaryMsg(resp)
 	c.close()
 	return err

internal/msggateway/hub_server.go

@@ -35,7 +35,7 @@ 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.ZookeeperConfig, &conf.MsgGateway.Prometheus, conf.MsgGateway.ListenIP,
+	return startrpc.Start(ctx, &conf.Discovery, &conf.MsgGateway.Prometheus, conf.MsgGateway.ListenIP,
 		conf.MsgGateway.RPC.RegisterIP,
 		conf.MsgGateway.RPC.Ports, index,
 		conf.Share.RpcRegisterName.MessageGateway,

internal/msggateway/init.go

@@ -24,10 +24,10 @@ import (
 )
 
 type Config struct {
-	MsgGateway      config.MsgGateway
-	ZookeeperConfig config.ZooKeeper
-	Share           config.Share
-	WebhooksConfig  config.Webhooks
+	MsgGateway     config.MsgGateway
+	Share          config.Share
+	WebhooksConfig config.Webhooks
+	Discovery      config.Discovery
 }
 
 // Start run ws server.

internal/msggateway/n_ws_server.go

@@ -211,7 +211,8 @@ func (ws *WsServer) sendUserOnlineInfoToOtherNode(ctx context.Context, client *C
 
 	// Online push user online message to other node
 	for _, v := range conns {
-		v := v // safe closure var
+		v := v
+		log.ZDebug(ctx, " sendUserOnlineInfoToOtherNode conn ", "target", v.Target())
 		if v.Target() == ws.disCov.GetSelfConnTarget() {
 			log.ZDebug(ctx, "Filter out this node", "node", v.Target())
 			continue
@@ -267,7 +268,9 @@ func (ws *WsServer) registerClient(client *Client) {
 	}
 
 	wg := sync.WaitGroup{}
-	if ws.msgGatewayConfig.Share.Env == "zookeeper" {
+	log.ZDebug(client.ctx, "ws.msgGatewayConfig.Discovery.Enable", ws.msgGatewayConfig.Discovery.Enable)
+
+	if ws.msgGatewayConfig.Discovery.Enable != "k8s" {
 		wg.Add(1)
 		go func() {
 			defer wg.Done()

internal/msgtransfer/init.go

@@ -56,13 +56,13 @@ type MsgTransfer struct {
 }
 
 type Config struct {
-	MsgTransfer     config.MsgTransfer
-	RedisConfig     config.Redis
-	MongodbConfig   config.Mongo
-	KafkaConfig     config.Kafka
-	ZookeeperConfig config.ZooKeeper
-	Share           config.Share
-	WebhooksConfig  config.Webhooks
+	MsgTransfer    config.MsgTransfer
+	RedisConfig    config.Redis
+	MongodbConfig  config.Mongo
+	KafkaConfig    config.Kafka
+	Share          config.Share
+	WebhooksConfig config.Webhooks
+	Discovery      config.Discovery
 }
 
 func Start(ctx context.Context, index int, config *Config) error {
@@ -76,7 +76,7 @@ func Start(ctx context.Context, index int, config *Config) error {
 	if err != nil {
 		return err
 	}
-	client, err := kdisc.NewDiscoveryRegister(&config.ZookeeperConfig, &config.Share)
+	client, err := kdisc.NewDiscoveryRegister(&config.Discovery, &config.Share)
 	if err != nil {
 		return err
 	}

internal/push/onlinepusher.go

@@ -12,11 +12,6 @@ import (
 	"sync"
 )
 
-const (
-	KUBERNETES = "k8s"
-	ZOOKEEPER  = "zookeeper"
-)
-
 type OnlinePusher interface {
 	GetConnsAndOnlinePush(ctx context.Context, msg *sdkws.MsgData,
 		pushToUserIDs []string) (wsResults []*msggateway.SingleMsgToUserResults, err error)
@@ -42,10 +37,12 @@ func (u emptyOnlinePUsher) GetOnlinePushFailedUserIDs(ctx context.Context, msg *
 }
 
 func NewOnlinePusher(disCov discovery.SvcDiscoveryRegistry, config *Config) OnlinePusher {
-	switch config.Share.Env {
-	case KUBERNETES:
+	switch config.Discovery.Enable {
+	case "k8s":
 		return NewK8sStaticConsistentHash(disCov, config)
-	case ZOOKEEPER:
+	case "zookeeper":
+		return NewDefaultAllNode(disCov, config)
+	case "etcd":
 		return NewDefaultAllNode(disCov, config)
 	default:
 		return newEmptyOnlinePUsher()
@@ -64,7 +61,12 @@ func NewDefaultAllNode(disCov discovery.SvcDiscoveryRegistry, config *Config) *D
 func (d *DefaultAllNode) GetConnsAndOnlinePush(ctx context.Context, msg *sdkws.MsgData,
 	pushToUserIDs []string) (wsResults []*msggateway.SingleMsgToUserResults, err error) {
 	conns, err := d.disCov.GetConns(ctx, d.config.Share.RpcRegisterName.MessageGateway)
-	log.ZDebug(ctx, "get gateway conn", "conn length", len(conns))
+	if len(conns) == 0 {
+		log.ZWarn(ctx, "get gateway conn 0 ", nil)
+	} else {
+		log.ZDebug(ctx, "get gateway conn", "conn length", len(conns))
+	}
+
 	if err != nil {
 		return nil, err
 	}
@@ -85,10 +87,12 @@ func (d *DefaultAllNode) GetConnsAndOnlinePush(ctx context.Context, msg *sdkws.M
 	// Online push message
 	for _, conn := range conns {
 		conn := conn // loop var safe
+		ctx := ctx
 		wg.Go(func() error {
 			msgClient := msggateway.NewMsgGatewayClient(conn)
 			reply, err := msgClient.SuperGroupOnlineBatchPushOneMsg(ctx, input)
 			if err != nil {
+				log.ZError(ctx, "SuperGroupOnlineBatchPushOneMsg ", err, "req:", input.String())
 				return nil
 			}
 

internal/push/push.go

@@ -24,11 +24,11 @@ type Config struct {
 	RedisConfig        config.Redis
 	MongodbConfig      config.Mongo
 	KafkaConfig        config.Kafka
-	ZookeeperConfig    config.ZooKeeper
 	NotificationConfig config.Notification
 	Share              config.Share
 	WebhooksConfig     config.Webhooks
 	LocalCacheConfig   config.LocalCache
+	Discovery          config.Discovery
 }
 
 func (p pushServer) PushMsg(ctx context.Context, req *pbpush.PushMsgReq) (*pbpush.PushMsgResp, error) {

internal/rpc/auth/auth.go

@@ -45,10 +45,10 @@ type authServer struct {
 }
 
 type Config struct {
-	RpcConfig       config.Auth
-	RedisConfig     config.Redis
-	ZookeeperConfig config.ZooKeeper
-	Share           config.Share
+	RpcConfig   config.Auth
+	RedisConfig config.Redis
+	Share       config.Share
+	Discovery   config.Discovery
 }
 
 func Start(ctx context.Context, config *Config, client discovery.SvcDiscoveryRegistry, server *grpc.Server) error {

internal/rpc/conversation/conversaion.go

@@ -51,10 +51,10 @@ type Config struct {
 	RpcConfig          config.Conversation
 	RedisConfig        config.Redis
 	MongodbConfig      config.Mongo
-	ZookeeperConfig    config.ZooKeeper
 	NotificationConfig config.Notification
 	Share              config.Share
 	LocalCacheConfig   config.LocalCache
+	Discovery          config.Discovery
 }
 
 func Start(ctx context.Context, config *Config, client discovery.SvcDiscoveryRegistry, server *grpc.Server) error {

internal/rpc/friend/friend.go

@@ -50,14 +50,15 @@ type friendServer struct {
 }
 
 type Config struct {
-	RpcConfig          config.Friend
-	RedisConfig        config.Redis
-	MongodbConfig      config.Mongo
-	ZookeeperConfig    config.ZooKeeper
+	RpcConfig     config.Friend
+	RedisConfig   config.Redis
+	MongodbConfig config.Mongo
+	//ZookeeperConfig    config.ZooKeeper
 	NotificationConfig config.Notification
 	Share              config.Share
 	WebhooksConfig     config.Webhooks
 	LocalCacheConfig   config.LocalCache
+	Discovery          config.Discovery
 }
 
 func Start(ctx context.Context, config *Config, client discovery.SvcDiscoveryRegistry, server *grpc.Server) error {

internal/rpc/group/group.go

@@ -68,11 +68,11 @@ type Config struct {
 	RpcConfig          config.Group
 	RedisConfig        config.Redis
 	MongodbConfig      config.Mongo
-	ZookeeperConfig    config.ZooKeeper
 	NotificationConfig config.Notification
 	Share              config.Share
 	WebhooksConfig     config.Webhooks
 	LocalCacheConfig   config.LocalCache
+	Discovery          config.Discovery
 }
 
 func Start(ctx context.Context, config *Config, client discovery.SvcDiscoveryRegistry, server *grpc.Server) error {

internal/rpc/msg/server.go

@@ -59,11 +59,11 @@ type (
 		RedisConfig        config.Redis
 		MongodbConfig      config.Mongo
 		KafkaConfig        config.Kafka
-		ZookeeperConfig    config.ZooKeeper
 		NotificationConfig config.Notification
 		Share              config.Share
 		WebhooksConfig     config.Webhooks
 		LocalCacheConfig   config.LocalCache
+		Discovery          config.Discovery
 	}
 )