feat: ver3 branch

Signed-off-by: kubbot & kubecub <3293172751ysy@gmail.com>

pkg/discoveryregistry/zookeeper/debug.go

@@ -1,47 +0,0 @@
-package zookeeper
-
-import (
-	"context"
-	"fmt"
-	"github.com/OpenIMSDK/Open-IM-Server/pkg/common/log"
-	"google.golang.org/grpc"
-	"strings"
-)
-
-func newClientConnInterface(cc grpc.ClientConnInterface) grpc.ClientConnInterface {
-	return &clientConnInterface{cc: cc}
-}
-
-type clientConnInterface struct {
-	cc grpc.ClientConnInterface
-}
-
-func (c *clientConnInterface) callOptionToString(opts []grpc.CallOption) string {
-	arr := make([]string, 0, len(opts)+1)
-	arr = append(arr, fmt.Sprintf("opts len: %d", len(opts)))
-	for i, opt := range opts {
-		arr = append(arr, fmt.Sprintf("[%d:%T]", i, opt))
-	}
-	return strings.Join(arr, ", ")
-}
-
-func (c *clientConnInterface) Invoke(ctx context.Context, method string, args interface{}, reply interface{}, opts ...grpc.CallOption) error {
-	log.ZDebug(ctx, "grpc.ClientConnInterface.Invoke in", "method", method, "args", args, "reply", reply, "opts", c.callOptionToString(opts))
-	if err := c.cc.Invoke(ctx, method, args, reply, opts...); err != nil {
-		log.ZError(ctx, "grpc.ClientConnInterface.Invoke error", err, "method", method, "args", args, "reply", reply)
-		return err
-	}
-	log.ZDebug(ctx, "grpc.ClientConnInterface.Invoke success", "method", method, "args", args, "reply", reply)
-	return nil
-}
-
-func (c *clientConnInterface) NewStream(ctx context.Context, desc *grpc.StreamDesc, method string, opts ...grpc.CallOption) (grpc.ClientStream, error) {
-	log.ZDebug(ctx, "grpc.ClientConnInterface.NewStream in", "desc", desc, "method", method, "opts", c.callOptionToString(opts))
-	cs, err := c.cc.NewStream(ctx, desc, method, opts...)
-	if err != nil {
-		log.ZError(ctx, "grpc.ClientConnInterface.NewStream error", err, "desc", desc, "method", method, "opts", len(opts))
-		return nil, err
-	}
-	log.ZDebug(ctx, "grpc.ClientConnInterface.NewStream success", "desc", desc, "method", method, "opts", len(opts))
-	return cs, nil
-}

pkg/discoveryregistry/zookeeper/discover.go

@@ -5,8 +5,8 @@ import (
 	"fmt"
 	"io"
 	"strings"
-	"sync"
 
+	"github.com/OpenIMSDK/Open-IM-Server/pkg/common/log"
 	"github.com/pkg/errors"
 
 	"github.com/go-zookeeper/zk"
@@ -17,7 +17,7 @@ import (
 var ErrConnIsNil = errors.New("conn is nil")
 var ErrConnIsNilButLocalNotNil = errors.New("conn is nil, but local is not nil")
 
-func (s *ZkClient) watch(wg *sync.WaitGroup) {
+func (s *ZkClient) watch() {
 	for {
 		event := <-s.eventChan
 		switch event.Type {
@@ -27,12 +27,9 @@ func (s *ZkClient) watch(wg *sync.WaitGroup) {
 			s.logger.Printf("zk event: %s", event.Path)
 			l := strings.Split(event.Path, "/")
 			if len(l) > 1 {
+				serviceName := l[len(l)-1]
 				s.lock.Lock()
-				rpcName := l[len(l)-1]
-				s.flushResolver(rpcName)
-				if len(s.localConns[rpcName]) != 0 {
-					delete(s.localConns, rpcName)
-				}
+				s.flushResolverAndDeleteLocal(serviceName)
 				s.lock.Unlock()
 			}
 			s.logger.Printf("zk event handle success: %s", event.Path)
@@ -47,27 +44,26 @@ func (s *ZkClient) watch(wg *sync.WaitGroup) {
 
 func (s *ZkClient) GetConnsRemote(serviceName string) (conns []resolver.Address, err error) {
 	path := s.getPath(serviceName)
-	childNodes, _, err := s.conn.Children(path)
-	if err != nil {
-		return nil, errors.Wrap(err, "get children error")
-	}
-	for _, child := range childNodes {
-		fullPath := path + "/" + child
-		data, _, err := s.conn.Get(fullPath)
-		if err != nil {
-			if err == zk.ErrNoNode {
-				return nil, errors.Wrap(err, "this is zk ErrNoNode")
-			}
-			return nil, errors.Wrap(err, "get children error")
-		}
-		conns = append(conns, resolver.Address{Addr: string(data), ServerName: serviceName})
-	}
-	_, _, _, err = s.conn.ChildrenW(s.getPath(serviceName))
+	_, _, _, err = s.conn.ChildrenW(path)
 	if err != nil {
 		return nil, errors.Wrap(err, "children watch error")
 	}
-	if len(conns) == 0 {
-		return nil, fmt.Errorf("no conn for service %s, grpc server may not exist, local conn is %v, please check zookeeper server %v, path: %s", serviceName, s.localConns, s.zkServers, s.zkRoot)
+	childNodes, _, err := s.conn.Children(path)
+	if err != nil {
+		return nil, errors.Wrap(err, "get children error")
+	} else {
+		for _, child := range childNodes {
+			fullPath := path + "/" + child
+			data, _, err := s.conn.Get(fullPath)
+			if err != nil {
+				if err == zk.ErrNoNode {
+					return nil, errors.Wrap(err, "this is zk ErrNoNode")
+				}
+				return nil, errors.Wrap(err, "get children error")
+			}
+			log.ZDebug(context.Background(), "get conns from remote", "conn", string(data))
+			conns = append(conns, resolver.Address{Addr: string(data), ServerName: serviceName})
+		}
 	}
 	return conns, nil
 }
@@ -75,7 +71,6 @@ func (s *ZkClient) GetConnsRemote(serviceName string) (conns []resolver.Address,
 func (s *ZkClient) GetConns(ctx context.Context, serviceName string, opts ...grpc.DialOption) ([]grpc.ClientConnInterface, error) {
 	s.logger.Printf("get conns from client, serviceName: %s", serviceName)
 	s.lock.Lock()
-	defer s.lock.Unlock()
 	opts = append(s.options, opts...)
 	conns := s.localConns[serviceName]
 	if len(conns) == 0 {
@@ -83,10 +78,15 @@ func (s *ZkClient) GetConns(ctx context.Context, serviceName string, opts ...grp
 		s.logger.Printf("get conns from zk remote, serviceName: %s", serviceName)
 		conns, err = s.GetConnsRemote(serviceName)
 		if err != nil {
+			s.lock.Unlock()
 			return nil, err
 		}
+		if len(conns) == 0 {
+			return nil, fmt.Errorf("no conn for service %s, grpc server may not exist, local conn is %v, please check zookeeper server %v, path: %s", serviceName, s.localConns, s.zkServers, s.zkRoot)
+		}
 		s.localConns[serviceName] = conns
 	}
+	s.lock.Unlock()
 	var ret []grpc.ClientConnInterface
 	s.logger.Printf("get conns from zk success, serviceName: %s", serviceName)
 	for _, conn := range conns {
@@ -94,7 +94,7 @@ func (s *ZkClient) GetConns(ctx context.Context, serviceName string, opts ...grp
 		if err != nil {
 			return nil, errors.Wrap(err, fmt.Sprintf("conns dialContext error, conn: %s", conn.Addr))
 		}
-		ret = append(ret, newClientConnInterface(cc))
+		ret = append(ret, cc)
 	}
 	s.logger.Printf("dial ctx success, serviceName: %s", serviceName)
 	return ret, nil
@@ -102,11 +102,8 @@ func (s *ZkClient) GetConns(ctx context.Context, serviceName string, opts ...grp
 
 func (s *ZkClient) GetConn(ctx context.Context, serviceName string, opts ...grpc.DialOption) (grpc.ClientConnInterface, error) {
 	newOpts := append(s.options, grpc.WithDefaultServiceConfig(fmt.Sprintf(`{"LoadBalancingPolicy": "%s"}`, s.balancerName)))
-	cc, err := grpc.DialContext(ctx, fmt.Sprintf("%s:///%s", s.scheme, serviceName), append(newOpts, opts...)...)
-	if err != nil {
-		return nil, err
-	}
-	return newClientConnInterface(cc), nil
+	s.logger.Printf("get conn from client, serviceName: %s", serviceName)
+	return grpc.DialContext(ctx, fmt.Sprintf("%s:///%s", s.scheme, serviceName), append(newOpts, opts...)...)
 }
 
 func (s *ZkClient) CloseConn(conn grpc.ClientConnInterface) {

pkg/discoveryregistry/zookeeper/register.go

@@ -8,6 +8,15 @@ import (
 	"google.golang.org/grpc/resolver"
 )
 
+func (s *ZkClient) CreateRpcRootNodes(serviceNames []string) error {
+	for _, serviceName := range serviceNames {
+		if err := s.ensureName(serviceName); err != nil && err != zk.ErrNodeExists {
+			return err
+		}
+	}
+	return nil
+}
+
 func (s *ZkClient) Register(rpcRegisterName, host string, port int, opts ...grpc.DialOption) error {
 	if err := s.ensureName(rpcRegisterName); err != nil {
 		return err

pkg/discoveryregistry/zookeeper/resolver.go

@@ -17,8 +17,8 @@ type Resolver struct {
 	getConnsRemote func(serviceName string) (conns []resolver.Address, err error)
 }
 
-func (r *Resolver) ResolveNow(o resolver.ResolveNowOptions) {
-	log.ZDebug(context.Background(), "start resolve now", "target", r.target)
+func (r *Resolver) ResolveNowZK(o resolver.ResolveNowOptions) {
+	log.ZDebug(context.Background(), "start resolve now", "target", r.target, "cc", r.cc.UpdateState, "serviceName", strings.TrimLeft(r.target.URL.Path, "/"))
 	newConns, err := r.getConnsRemote(strings.TrimLeft(r.target.URL.Path, "/"))
 	if err != nil {
 		log.ZError(context.Background(), "resolve now error", err, "target", r.target)
@@ -26,24 +26,30 @@ func (r *Resolver) ResolveNow(o resolver.ResolveNowOptions) {
 	}
 	r.addrs = newConns
 	if err := r.cc.UpdateState(resolver.State{Addresses: newConns}); err != nil {
-		log.ZError(context.Background(), "UpdateState error", err, "conns", newConns)
+		log.ZError(context.Background(), "UpdateState error, conns is nil from svr", err, "conns", newConns, "zk path", r.target.URL.Path)
+		return
 	}
 	log.ZDebug(context.Background(), "resolve now finished", "target", r.target, "conns", r.addrs)
 }
 
+func (r *Resolver) ResolveNow(o resolver.ResolveNowOptions) {}
+
 func (s *Resolver) Close() {}
 
 func (s *ZkClient) Build(target resolver.Target, cc resolver.ClientConn, opts resolver.BuildOptions) (resolver.Resolver, error) {
-	s.logger.Printf("build resolver: %+v, cc: %+v", target, cc)
+	s.logger.Printf("build resolver: %+v, cc: %+v", target, cc.UpdateState)
+	// log.ZDebug(context.Background(), "build resolver start", "target", target, "cc", cc.UpdateState)
 	r := &Resolver{}
 	r.target = target
 	r.cc = cc
 	r.getConnsRemote = s.GetConnsRemote
-	r.ResolveNow(resolver.ResolveNowOptions{})
+	r.ResolveNowZK(resolver.ResolveNowOptions{})
 	s.lock.Lock()
 	defer s.lock.Unlock()
-	s.resolvers[strings.TrimLeft(target.URL.Path, "/")] = r
-	s.logger.Printf("build resolver finished: %+v, cc: %+v", target, cc)
+	serviceName := strings.TrimLeft(target.URL.Path, "/")
+	s.resolvers[serviceName] = r
+	s.logger.Printf("build resolver finished: %+v, cc: %+v, key: %s", target, cc.UpdateState, serviceName)
+	// log.ZDebug(context.Background(), "build resolver finished", "target", target, "cc", cc.UpdateState, "serviceName", serviceName)
 	return r, nil
 }
 

pkg/discoveryregistry/zookeeper/zk.go

@@ -38,7 +38,7 @@ type ZkClient struct {
 	options []grpc.DialOption
 
 	resolvers    map[string]*Resolver
-	localConns   map[string][]resolver.Address //msg gateway
+	localConns   map[string][]resolver.Address
 	balancerName string
 
 	logger Logger
@@ -114,9 +114,8 @@ func NewClient(zkServers []string, zkRoot string, options ...ZkOption) (*ZkClien
 		return nil, err
 	}
 	resolver.Register(client)
-	var wg sync.WaitGroup
-	go client.refresh(&wg)
-	go client.watch(&wg)
+	go client.refresh()
+	go client.watch()
 	return client, nil
 }
 
@@ -138,7 +137,7 @@ func (s *ZkClient) ensureAndCreate(node string) error {
 	return nil
 }
 
-func (s *ZkClient) refresh(wg *sync.WaitGroup) {
+func (s *ZkClient) refresh() {
 	for range s.ticker.C {
 		s.logger.Printf("refresh local conns")
 		s.lock.Lock()
@@ -149,16 +148,20 @@ func (s *ZkClient) refresh(wg *sync.WaitGroup) {
 			delete(s.localConns, rpcName)
 		}
 		s.lock.Unlock()
+		s.logger.Printf("refresh local conns success")
 	}
+}
 
+func (s *ZkClient) flushResolverAndDeleteLocal(serviceName string) {
+	s.logger.Printf("start flush %s", serviceName)
+	s.flushResolver(serviceName)
+	delete(s.localConns, serviceName)
 }
 
 func (s *ZkClient) flushResolver(serviceName string) {
-	s.logger.Printf("start flush")
 	r, ok := s.resolvers[serviceName]
 	if ok {
-		r.ResolveNow(resolver.ResolveNowOptions{})
-		s.resolvers[serviceName] = r
+		r.ResolveNowZK(resolver.ResolveNowOptions{})
 	}
 }
 
@@ -197,8 +200,3 @@ func (s *ZkClient) AddOption(opts ...grpc.DialOption) {
 func (s *ZkClient) GetClientLocalConns() map[string][]resolver.Address {
 	return s.localConns
 }
-
-type FakeLock struct{}
-
-func (s *FakeLock) Lock()   {}
-func (s *FakeLock) Unlock() {}