refactor zk dynamic configuration listener

cluster/router/v3router/router_chain.go

@@ -35,6 +35,7 @@ import (
  "dubbo.apache.org/dubbo-go/v3/config"
  "dubbo.apache.org/dubbo-go/v3/config_center"
  "dubbo.apache.org/dubbo-go/v3/protocol"
+ "dubbo.apache.org/dubbo-go/v3/remoting"
 )
 
 // RouterChain contains all uniform router logic
@@ -46,31 +47,36 @@ type RouterChain struct {
 
 // nolint
 func NewUniformRouterChain() (router.PriorityRouter, error) {
- // 1. add mesh route listener
+ // 1. Add mesh route listener
  r := &RouterChain{}
  rootConfig := config.GetRootConfig()
  dynamicConfiguration := conf.GetEnvInstance().GetDynamicConfiguration()
  if dynamicConfiguration == nil {
- logger.Infof("[Mesh Router] Config center does not start, please check if the configuration center has been properly configured in dubbogo.yml")
+ logger.Infof("[NewUniformRouterChain] Config center does not start, please check if the configuration center has been properly configured in dubbogo.yml")
  return nil, nil
  }
- dynamicConfiguration.AddListener(rootConfig.Application.Name, r)
 
- // 2. try to get mesh route configuration, default key is "dubbo.io.MESHAPPRULE" with group "dubbo"
+ // 2. Try to get mesh rules configuration, default key is "dubbo.io.MESHAPPRULE" with group "dubbo"
  key := rootConfig.Application.Name + constant.MeshRouteSuffix
+ group := rootConfig.ConfigCenter.Group
+ if group == "" {
+ group = constant.Dubbo
+ }
+ dynamicConfiguration.AddListener(group+constant.PathSeparator+key, r)
  meshRouteValue, err := dynamicConfiguration.GetProperties(key, config_center.WithGroup(rootConfig.ConfigCenter.Group))
  if err != nil {
- // the mesh route may not be initialized now
- logger.Warnf("Can not get mesh route for key=%s, error=%v", key, err)
+ // The mesh rules may not be initialized now
+ logger.Warnf("[NewUniformRouterChain]Can not get mesh rules for group=%s, key=%s, error=%+v", rootConfig.ConfigCenter.Group, key, err)
  return r, nil
  }
- logger.Debugf("Successfully get mesh route:%s", meshRouteValue)
+ logger.Debugf("[NewUniformRouterChain]Successfully get mesh rules:%s", meshRouteValue)
  routes, err := parseRoute(meshRouteValue)
  if err != nil {
- logger.Warnf("Parse mesh route failed, error=%v", err)
+ logger.Warnf("[NewUniformRouterChain]Parse mesh rules failed, error=%+v", err)
  return nil, err
  }
  r.routers = routes
+ logger.Infof("[NewUniformRouterChain]Successfully init mesh rules with:\n%s", meshRouteValue)
  return r, nil
 }
 
@@ -84,13 +90,19 @@ func (r *RouterChain) Route(invokers []protocol.Invoker, url *common.URL, invoca
 
 // Process process route config change event
 func (r *RouterChain) Process(event *config_center.ConfigChangeEvent) {
- logger.Debugf("RouteChain process event:\n%+v", event)
+ logger.Infof("[RouteChain]Process config change event:%+v", event)
+ if event.ConfigType == remoting.EventTypeDel {
+ r.routers = nil
+ return
+ }
  routers, err := parseRoute(event.Value.(string))
  if err != nil {
+ logger.Warnf("[RouteChain]Parse new mesh route config error, %+v "+
+ "and we will use the original mesh rule configuration.", err)
  return
  }
  r.routers = routers
- // todo delete router
+ logger.Infof("[RouteChain]Parse Mesh Rule Success.")
 }
 
 // Name get name of ConnCheckerRouter
@@ -108,7 +120,7 @@ func (r *RouterChain) URL() *common.URL {
  return nil
 }
 
-// parseFromConfigToRouters parse virtualService and destinationRule yaml file bytes to target router list
+// Deprecated parseFromConfigToRouters parse virtualService and destinationRule yaml file bytes to target router list
 func parseFromConfigToRouters(virtualServiceConfig, destinationRuleConfig []byte) ([]*UniformRouter, error) {
  var virtualServiceConfigList []*config.VirtualServiceConfig
  destRuleConfigsMap := make(map[string]map[string]map[string]string)

common/constant/default.go

@@ -27,11 +27,8 @@ const (
 )
 
 const (
- DefaultWeight = 100 //
- DefaultWarmup = 10 * 60 // in java here is 10*60*1000 because of System.currentTimeMillis() is measured in milliseconds & in go time.Unix() is second
-)
-
-const (
+ DefaultWeight = 100 //
+ DefaultWarmup = 10 * 60 // in java here is 10*60*1000 because of System.currentTimeMillis() is measured in milliseconds & in go time.Unix() is second
  DefaultLoadBalance = "random"
  DefaultRetries = "2"
  DefaultRetriesInt = 2

common/constant/key.go

@@ -49,6 +49,7 @@ const (
  PortKey = "port"
  ProtocolKey = "protocol"
  PathSeparator = "/"
+ DotSeparator = "."
  CommaSeparator = ","
  SslEnabledKey = "ssl-enabled"
  // ParamsTypeKey key used in pass through invoker factory, to define param type

config/consumer_config.go

@@ -83,7 +83,7 @@ func (cc *ConsumerConfig) Init(rc *RootConfig) error {
  // try to use interface name defined by pb
  triplePBService, ok := reference.(common.TriplePBService)
  if !ok {
- logger.Errorf("Dubbogo cannot get interface name with reference = %s."+
+ logger.Errorf("Dubbo-go cannot get interface name with reference = %s."+
  "Please run the command 'go install github.com/dubbogo/tools/cmd/protoc-gen-go-triple@latest' to get the latest "+
  "protoc-gen-go-triple, and then re-generate your pb file again by this tool."+
  "If you are not using pb serialization, please set 'interfaceName' field in reference config to let dubbogo get the interface name.", key)

config_center/zookeeper/impl.go

@@ -20,6 +20,7 @@ package zookeeper
 import (
  "encoding/base64"
  "strconv"
+ "strings"
  "sync"
 )
 
@@ -65,8 +66,9 @@ type zookeeperDynamicConfiguration struct {
 
 func newZookeeperDynamicConfiguration(url *common.URL) (*zookeeperDynamicConfiguration, error) {
  c := &zookeeperDynamicConfiguration{
- url: url,
- rootPath: "/" + url.GetParam(constant.ConfigNamespaceKey, config_center.DefaultGroup) + "/config",
+ url: url,
+ // TODO adapt config center config
+ rootPath: "/dubbo/config",
  }
  logger.Infof("[Zookeeper ConfigCenter] New Zookeeper ConfigCenter with Configuration: %+v, url = %+v", c, c.GetURL())
  if v, ok := config.GetRootConfig().ConfigCenter.Params["base64"]; ok {
@@ -93,13 +95,26 @@ func newZookeeperDynamicConfiguration(url *common.URL) (*zookeeperDynamicConfigu
 
  // Start listener
  c.listener = zookeeper.NewZkEventListener(c.client)
- c.cacheListener = NewCacheListener(c.rootPath)
- c.listener.ListenServiceEvent(url, c.rootPath, c.cacheListener)
+ c.cacheListener = NewCacheListener(c.rootPath, c.listener)
+ c.listener.ListenConfigurationEvent(c.rootPath, c.cacheListener)
  return c, nil
 }
 
-func (c *zookeeperDynamicConfiguration) AddListener(key string, listener config_center.ConfigurationListener, opions ...config_center.Option) {
- c.cacheListener.AddListener(key, listener)
+// AddListener add listener for key
+// TODO this method should has a parameter 'group', and it does not now, so we should concat group and key with '/' manually
+func (c *zookeeperDynamicConfiguration) AddListener(key string, listener config_center.ConfigurationListener, options ...config_center.Option) {
+ qualifiedKey := buildPath(c.rootPath, key)
+ c.cacheListener.AddListener(qualifiedKey, listener)
+}
+
+// buildPath build path and format
+func buildPath(rootPath, subPath string) string {
+ path := strings.TrimRight(rootPath+pathSeparator+subPath, pathSeparator)
+ if !strings.HasPrefix(path, pathSeparator) {
+ path = pathSeparator + path
+ }
+ path = strings.ReplaceAll(path, "//", "/")
+ return path
 }
 
 func (c *zookeeperDynamicConfiguration) RemoveListener(key string, listener config_center.ConfigurationListener, opions ...config_center.Option) {
@@ -118,7 +133,7 @@ func (c *zookeeperDynamicConfiguration) GetProperties(key string, opts ...config
  if len(tmpOpts.Group) != 0 {
  key = tmpOpts.Group + "/" + key
  } else {
- key = c.GetURL().GetParam(constant.ConfigNamespaceKey, config_center.DefaultGroup)
+ key = c.GetURL().GetParam(constant.ConfigNamespaceKey, config_center.DefaultGroup) + "/" + key
  }
  content, _, err := c.client.GetContent(c.rootPath + "/" + key)
  if err != nil {

config_center/zookeeper/listener.go

@@ -24,27 +24,33 @@ import (
 
 import (
  "dubbo.apache.org/dubbo-go/v3/common/constant"
- "dubbo.apache.org/dubbo-go/v3/common/logger"
- "dubbo.apache.org/dubbo-go/v3/config"
  "dubbo.apache.org/dubbo-go/v3/config_center"
  "dubbo.apache.org/dubbo-go/v3/remoting"
+ "dubbo.apache.org/dubbo-go/v3/remoting/zookeeper"
 )
 
 // CacheListener defines keyListeners and rootPath
 type CacheListener struct {
- keyListeners sync.Map
- rootPath string
+ // key is zkNode Path and value is set of listeners
+ keyListeners sync.Map
+ zkEventListener *zookeeper.ZkEventListener
+ rootPath string
 }
 
 // NewCacheListener creates a new CacheListener
-func NewCacheListener(rootPath string) *CacheListener {
- return &CacheListener{rootPath: rootPath}
+func NewCacheListener(rootPath string, listener *zookeeper.ZkEventListener) *CacheListener {
+ return &CacheListener{zkEventListener: listener, rootPath: rootPath}
 }
 
 // AddListener will add a listener if loaded
 func (l *CacheListener) AddListener(key string, listener config_center.ConfigurationListener) {
+ // FIXME do not use Client.ExistW, cause it has a bug(can not watch zk node that do not exist)
+ _, _, _, err := l.zkEventListener.Client.Conn.ExistsW(key)
  // reference from https://stackoverflow.com/questions/34018908/golang-why-dont-we-have-a-set-datastructure
  // make a map[your type]struct{} like set in java
+ if err != nil {
+ return
+ }
  listeners, loaded := l.keyListeners.LoadOrStore(key, map[config_center.ConfigurationListener]struct{}{listener: {}})
  if loaded {
  listeners.(map[config_center.ConfigurationListener]struct{})[listener] = struct{}{}
@@ -62,36 +68,28 @@ func (l *CacheListener) RemoveListener(key string, listener config_center.Config
 
 // DataChange changes all listeners' event
 func (l *CacheListener) DataChange(event remoting.Event) bool {
+ changeType := event.Action
  if event.Content == "" {
- // meanings new node
- return true
- }
- var key string
- // TODO use common way
- if strings.HasSuffix(event.Path, constant.MeshRouteSuffix) {
- key = config.GetRootConfig().Application.Name
- } else {
- key = l.pathToKey(event.Path)
+ changeType = remoting.EventTypeDel
  }
- if key != "" {
- if listeners, ok := l.keyListeners.Load(key); ok {
- for listener := range listeners.(map[config_center.ConfigurationListener]struct{}) {
- listener.Process(&config_center.ConfigChangeEvent{Key: key, Value: event.Content, ConfigType: event.Action})
- }
- return true
+
+ if listeners, ok := l.keyListeners.Load(event.Path); ok {
+ for listener := range listeners.(map[config_center.ConfigurationListener]struct{}) {
+ listener.Process(&config_center.ConfigChangeEvent{
+ Key: l.pathToKey(event.Path),
+ Value: event.Content,
+ ConfigType: changeType,
+ })
  }
+ return true
  }
  return false
 }
 
 func (l *CacheListener) pathToKey(path string) string {
- key := strings.Replace(strings.Replace(path, l.rootPath+"/", "", -1), "/", ".", -1)
- if strings.HasSuffix(key, constant.ConfiguratorSuffix) ||
- strings.HasSuffix(key, constant.TagRouterRuleSuffix) ||
- strings.HasSuffix(key, constant.ConditionRouterRuleSuffix) {
- // governance config, so we remove the "dubbo." prefix
- key = key[strings.Index(key, ".")+1:]
+ if len(path) == 0 {
+ return path
  }
- logger.Debugf("pathToKey path:%s, key:%s\n", path, key)
- return key
+ groupKey := strings.Replace(strings.Replace(path, l.rootPath+constant.PathSeparator, "", -1), constant.PathSeparator, constant.DotSeparator, -1)
+ return groupKey[strings.Index(groupKey, constant.DotSeparator)+1:]
 }

remoting/listener.go

@@ -23,7 +23,7 @@ import (
 
 // DataListener defines common data listener interface
 type DataListener interface {
- DataChange(eventType Event) bool // bool is return for interface implement is interesting
+ DataChange(event Event) bool // bool is return for interface implement is interesting
 }
 
 //////////////////////////////////////////

remoting/zookeeper/client.go

@@ -52,7 +52,7 @@ func ValidateZookeeperClient(container ZkClientFacade, zkName string) error {
  timeout := url.GetParamDuration(constant.ConfigTimeoutKey, constant.DefaultRegTimeout)
 
  zkAddresses := strings.Split(url.Location, ",")
- logger.Infof("[Zookeeper Client] New zookeeper client with name = %s, zkAddress = %s, timeout = %d", zkName, url.Location, timeout.String())
+ logger.Infof("[Zookeeper Client] New zookeeper client with name = %s, zkAddress = %s, timeout = %s", zkName, url.Location, timeout.String())
  newClient, cltErr := gxzookeeper.NewZookeeperClient(zkName, zkAddresses, true, gxzookeeper.WithZkTimeOut(timeout))
  if cltErr != nil {
  logger.Warnf("newZookeeperClient(name{%s}, zk address{%v}, timeout{%d}) = error{%v}",