Merge pull request #1010 from wenxuwan/fix_zk_too_many_tcp_conn

Fix:zk too many tcp conn

cluster/router/chain/chain_test.go

@@ -25,6 +25,8 @@ import (
 )
 
 import (
+ zk "github.com/dubbogo/go-zookeeper/zk"
+ gxzookeeper "github.com/dubbogo/gost/database/kv/zk"
  "github.com/stretchr/testify/assert"
 )
 
@@ -38,7 +40,6 @@ import (
  _ "github.com/apache/dubbo-go/config_center/zookeeper"
  "github.com/apache/dubbo-go/protocol"
  "github.com/apache/dubbo-go/protocol/invocation"
- "github.com/apache/dubbo-go/remoting/zookeeper"
 )
 
 const (
@@ -53,16 +54,19 @@ const (
  consumerFormat = "consumer://%s/com.foo.BarService"
  dubboForamt = "dubbo://%s:%d/com.foo.BarService"
  anyUrlFormat = "condition://%s/com.foo.BarService"
- zk  = "zookeeper"
+ zkName = "zookeeper"
  applicationKey = "test-condition"
  applicationField = "application"
  forceField = "force"
  forceValue = "true"
 )
 
+var zkCluster *zk.TestCluster
+
 func TestNewRouterChain(t *testing.T) {
- ts, z, _, err := zookeeper.NewMockZookeeperClient("test", 15*time.Second)
+ ts, z, _, err := gxzookeeper.NewMockZookeeperClient("test", 15*time.Second)
  assert.NoError(t, err)
+ zkCluster = ts
  err = z.Create(path)
  assert.NoError(t, err)
  testyml := `scope: application
@@ -77,12 +81,12 @@ conditions:
  _, err = z.Conn.Set(path, []byte(testyml), 0)
  assert.NoError(t, err)
  defer func() {
- _ = ts.Stop()
+ z.Delete(path)
  z.Close()
  }()
 
  zkUrl, _ := common.NewURL(fmt.Sprintf(zkFormat, localIP, ts.Servers[0].Port))
- configuration, err := extension.GetConfigCenterFactory(zk).GetDynamicConfiguration(zkUrl)
+ configuration, err := extension.GetConfigCenterFactory(zkName).GetDynamicConfiguration(zkUrl)
  config.GetEnvInstance().SetDynamicConfiguration(configuration)
 
  assert.Nil(t, err)
@@ -114,7 +118,7 @@ func TestNewRouterChainURLNil(t *testing.T) {
 }
 
 func TestRouterChainAddRouters(t *testing.T) {
- ts, z, _, err := zookeeper.NewMockZookeeperClient("test", 15*time.Second)
+ _, z, _, err := gxzookeeper.NewMockZookeeperClient("test", 15*time.Second, gxzookeeper.WithTestCluster(zkCluster))
  assert.NoError(t, err)
  err = z.Create(path)
  assert.NoError(t, err)
@@ -131,13 +135,12 @@ conditions:
  _, err = z.Conn.Set(path, []byte(testyml), 0)
  assert.NoError(t, err)
  defer func() {
- _ = ts.Stop()
- assert.NoError(t, err)
+ z.Delete(path)
  z.Close()
  }()
 
- zkUrl, _ := common.NewURL(fmt.Sprintf(zkFormat, localIP, ts.Servers[0].Port))
- configuration, err := extension.GetConfigCenterFactory(zk).GetDynamicConfiguration(zkUrl)
+ zkUrl, _ := common.NewURL(fmt.Sprintf(zkFormat, localIP, zkCluster.Servers[0].Port))
+ configuration, err := extension.GetConfigCenterFactory(zkName).GetDynamicConfiguration(zkUrl)
  assert.NoError(t, err)
  config.GetEnvInstance().SetDynamicConfiguration(configuration)
 
@@ -164,15 +167,10 @@ conditions:
 }
 
 func TestRouterChainRoute(t *testing.T) {
- ts, z, _, err := zookeeper.NewMockZookeeperClient("test", 15*time.Second)
- defer func() {
- err = ts.Stop()
- assert.NoError(t, err)
- z.Close()
- }()
-
+ ts, _, _, err := gxzookeeper.NewMockZookeeperClient("test", 15*time.Second, gxzookeeper.WithTestCluster(zkCluster))
+ assert.Nil(t, err)
  zkUrl, _ := common.NewURL(fmt.Sprintf(zkFormat, localIP, ts.Servers[0].Port))
- configuration, err := extension.GetConfigCenterFactory(zk).GetDynamicConfiguration(zkUrl)
+ configuration, err := extension.GetConfigCenterFactory(zkName).GetDynamicConfiguration(zkUrl)
  assert.NoError(t, err)
  config.GetEnvInstance().SetDynamicConfiguration(configuration)
 
@@ -197,7 +195,7 @@ func TestRouterChainRoute(t *testing.T) {
 }
 
 func TestRouterChainRouteAppRouter(t *testing.T) {
- ts, z, _, err := zookeeper.NewMockZookeeperClient("test", 15*time.Second)
+ ts, z, _, err := gxzookeeper.NewMockZookeeperClient("test", 15*time.Second, gxzookeeper.WithTestCluster(zkCluster))
  assert.NoError(t, err)
  err = z.Create(path)
  assert.NoError(t, err)
@@ -214,13 +212,12 @@ conditions:
  _, err = z.Conn.Set(path, []byte(testyml), 0)
  assert.NoError(t, err)
  defer func() {
- _ = ts.Stop()
- assert.NoError(t, err)
+ z.Delete(path)
  z.Close()
  }()
 
  zkUrl, _ := common.NewURL(fmt.Sprintf(zkFormat, localIP, ts.Servers[0].Port))
- configuration, err := extension.GetConfigCenterFactory(zk).GetDynamicConfiguration(zkUrl)
+ configuration, err := extension.GetConfigCenterFactory(zkName).GetDynamicConfiguration(zkUrl)
  assert.NoError(t, err)
  config.GetEnvInstance().SetDynamicConfiguration(configuration)
 
@@ -242,7 +239,7 @@ conditions:
 }
 
 func TestRouterChainRouteNoRoute(t *testing.T) {
- ts, z, _, err := zookeeper.NewMockZookeeperClient("test", 15*time.Second)
+ ts, z, _, err := gxzookeeper.NewMockZookeeperClient("test", 15*time.Second, gxzookeeper.WithTestCluster(zkCluster))
  assert.Nil(t, err)
  defer func() {
  _ = ts.Stop()
@@ -251,7 +248,7 @@ func TestRouterChainRouteNoRoute(t *testing.T) {
  }()
 
  zkUrl, _ := common.NewURL(fmt.Sprintf(zkFormat, localIP, ts.Servers[0].Port))
- configuration, err := extension.GetConfigCenterFactory(zk).GetDynamicConfiguration(zkUrl)
+ configuration, err := extension.GetConfigCenterFactory(zkName).GetDynamicConfiguration(zkUrl)
  config.GetEnvInstance().SetDynamicConfiguration(configuration)
  assert.Nil(t, err)
 

cluster/router/condition/app_router_test.go

@@ -24,6 +24,7 @@ import (
 )
 
 import (
+ gxzookeeper "github.com/dubbogo/gost/database/kv/zk"
  "github.com/stretchr/testify/assert"
 )
 
@@ -35,7 +36,6 @@ import (
  "github.com/apache/dubbo-go/config_center"
  _ "github.com/apache/dubbo-go/config_center/zookeeper"
  "github.com/apache/dubbo-go/remoting"
- "github.com/apache/dubbo-go/remoting/zookeeper"
 )
 
 const (
@@ -60,15 +60,14 @@ runtime: false
 conditions:
  - => host != 172.22.3.91
 `
- ts, z, _, err := zookeeper.NewMockZookeeperClient("test", 15*time.Second)
+ ts, z, _, err := gxzookeeper.NewMockZookeeperClient("test", 15*time.Second)
  assert.NoError(t, err)
  err = z.Create(routerPath)
  assert.NoError(t, err)
 
  _, err = z.Conn.Set(routerPath, []byte(testYML), 0)
  assert.NoError(t, err)
  defer func() {
- err = ts.Stop()
  assert.NoError(t, err)
  z.Close()
  }()
@@ -116,15 +115,14 @@ conditions:
  - => host != 172.22.3.91
  - host = 192.168.199.208 => host = 192.168.199.208
 `
- ts, z, _, err := zookeeper.NewMockZookeeperClient("test", 15*time.Second)
+ ts, z, _, err := gxzookeeper.NewMockZookeeperClient("test", 15*time.Second)
  assert.NoError(t, err)
  err = z.Create(routerPath)
  assert.NoError(t, err)
 
  _, err = z.Conn.Set(routerPath, []byte(testYML), 0)
  assert.NoError(t, err)
  defer func() {
- err = ts.Stop()
  assert.NoError(t, err)
  z.Close()
  }()
@@ -163,7 +161,7 @@ runtime: false
 conditions:
  - => host != 172.22.3.91
 `
- ts, z, _, err := zookeeper.NewMockZookeeperClient("test", 15*time.Second)
+ ts, z, _, err := gxzookeeper.NewMockZookeeperClient("test", 15*time.Second)
  assert.NoError(t, err)
  err = z.Create(routerPath)
  assert.NoError(t, err)

cluster/router/tag/tag_router_test.go

@@ -27,6 +27,7 @@ import (
 import (
  "github.com/RoaringBitmap/roaring"
  "github.com/dubbogo/go-zookeeper/zk"
+ gxzookeeper "github.com/dubbogo/gost/database/kv/zk"
  "github.com/stretchr/testify/assert"
  "github.com/stretchr/testify/suite"
 )
@@ -43,7 +44,6 @@ import (
  "github.com/apache/dubbo-go/protocol"
  "github.com/apache/dubbo-go/protocol/invocation"
  "github.com/apache/dubbo-go/remoting"
- "github.com/apache/dubbo-go/remoting/zookeeper"
 )
 
 const (
@@ -260,7 +260,7 @@ type DynamicTagRouter struct {
  //rule *RouterRule
 
  route *tagRouter
- zkClient *zookeeper.ZookeeperClient
+ zkClient *gxzookeeper.ZookeeperClient
  testCluster *zk.TestCluster
  invokers []protocol.Invoker
  url *common.URL
@@ -299,7 +299,7 @@ tags:
  - name: tag3
  addresses: ["127.0.0.1:20003", "127.0.0.1:20004"]
 `
- ts, z, _, err := zookeeper.NewMockZookeeperClient("test", 15*time.Second)
+ ts, z, _, err := gxzookeeper.NewMockZookeeperClient("test", 15*time.Second)
  suite.NoError(err)
  err = z.Create(routerPath)
  suite.NoError(err)
@@ -334,8 +334,6 @@ tags:
 
 func (suite *DynamicTagRouter) TearDownTest() {
  suite.zkClient.Close()
- err := suite.testCluster.Stop()
- suite.Nil(err)
 }
 
 func (suite *DynamicTagRouter) TestDynamicTagRouterSetByIPv4() {

config_center/zookeeper/impl.go

@@ -24,6 +24,7 @@ import (
 
 import (
  gxset "github.com/dubbogo/gost/container/set"
+ gxzookeeper "github.com/dubbogo/gost/database/kv/zk"
  perrors "github.com/pkg/errors"
 )
 
@@ -50,7 +51,7 @@ type zookeeperDynamicConfiguration struct {
  wg sync.WaitGroup
  cltLock sync.Mutex
  done chan struct{}
- client *zookeeper.ZookeeperClient
+ client *gxzookeeper.ZookeeperClient
 
  //listenerLock sync.Mutex
  listener *zookeeper.ZkEventListener
@@ -63,7 +64,7 @@ func newZookeeperDynamicConfiguration(url *common.URL) (*zookeeperDynamicConfigu
  url: url,
  rootPath: "/" + url.GetParam(constant.CONFIG_NAMESPACE_KEY, config_center.DEFAULT_GROUP) + "/config",
  }
- err := zookeeper.ValidateZookeeperClient(c, zookeeper.WithZkName(ZkClient))
+ err := zookeeper.ValidateZookeeperClient(c, ZkClient)
  if err != nil {
  logger.Errorf("zookeeper client start error ,error message is %v", err)
  return nil, err
@@ -163,11 +164,11 @@ func (c *zookeeperDynamicConfiguration) SetParser(p parser.ConfigurationParser)
  c.parser = p
 }
 
-func (c *zookeeperDynamicConfiguration) ZkClient() *zookeeper.ZookeeperClient {
+func (c *zookeeperDynamicConfiguration) ZkClient() *gxzookeeper.ZookeeperClient {
  return c.client
 }
 
-func (c *zookeeperDynamicConfiguration) SetZkClient(client *zookeeper.ZookeeperClient) {
+func (c *zookeeperDynamicConfiguration) SetZkClient(client *gxzookeeper.ZookeeperClient) {
  c.client = client
 }
 
@@ -206,10 +207,9 @@ func (c *zookeeperDynamicConfiguration) IsAvailable() bool {
 }
 
 func (c *zookeeperDynamicConfiguration) closeConfigs() {
+ logger.Infof("begin to close provider zk client")
  c.cltLock.Lock()
  defer c.cltLock.Unlock()
- logger.Infof("begin to close provider zk client")
- // Close the old client first to close the tmp node
  c.client.Close()
  c.client = nil
 }

config_center/zookeeper/impl_test.go

@@ -81,10 +81,10 @@ func initZkData(group string, t *testing.T) (*zk.TestCluster, *zookeeperDynamicC
  dubbo.service.com.ikurento.user.UserProvider.cluster=failover
 `
  if group != "" {
- err = zreg.client.Create(path.Join(zreg.rootPath, "dubbo", dubboPropertyFileName))
+ err = zreg.client.Create(path.Join(zreg.rootPath, group, dubboPropertyFileName))
  assert.NoError(t, err)
 
- _, err = zreg.client.Conn.Set(path.Join(zreg.rootPath, "dubbo", dubboPropertyFileName), []byte(data), 0)
+ _, err = zreg.client.Conn.Set(path.Join(zreg.rootPath, group, dubboPropertyFileName), []byte(data), 0)
  assert.NoError(t, err)
  } else {
  err = zreg.client.Create(path.Join(zreg.rootPath, dubboPropertyFileName))
@@ -100,6 +100,7 @@ func initZkData(group string, t *testing.T) (*zk.TestCluster, *zookeeperDynamicC
 func TestGetConfig(t *testing.T) {
  ts, reg := initZkData("dubbo", t)
  defer func() {
+ reg.client.Close()
  err := ts.Stop()
  assert.NoError(t, err)
  }()
@@ -122,11 +123,13 @@ func TestGetConfig(t *testing.T) {
 func TestAddListener(t *testing.T) {
  ts, reg := initZkData("", t)
  defer func() {
+ reg.client.Close()
  err := ts.Stop()
  assert.NoError(t, err)
  }()
  listener := &mockDataListener{}
  reg.AddListener(dubboPropertyFileName, listener)
+
  listener.wg.Add(1)
  data := `
  dubbo.consumer.request_timeout=3s
@@ -158,6 +161,7 @@ func TestAddListener(t *testing.T) {
 func TestRemoveListener(t *testing.T) {
  ts, reg := initZkData("", t)
  defer func() {
+ reg.client.Close()
  err := ts.Stop()
  assert.NoError(t, err)
  }()
@@ -197,19 +201,20 @@ func TestZookeeperDynamicConfigurationPublishConfig(t *testing.T) {
  value := "Test Data"
  customGroup := "Custom Group"
  key := "myKey"
- ts, zk := initZkData(config_center.DEFAULT_GROUP, t)
+ ts, reg := initZkData(config_center.DEFAULT_GROUP, t)
  defer func() {
+ reg.client.Close()
  err := ts.Stop()
  assert.NoError(t, err)
  }()
- err := zk.PublishConfig(key, customGroup, value)
+ err := reg.PublishConfig(key, customGroup, value)
  assert.Nil(t, err)
- result, err := zk.GetInternalProperty("myKey", config_center.WithGroup(customGroup))
+ result, err := reg.GetInternalProperty("myKey", config_center.WithGroup(customGroup))
  assert.Nil(t, err)
  assert.Equal(t, value, result)
 
  var keys *gxset.HashSet
- keys, err = zk.GetConfigKeysByGroup(customGroup)
+ keys, err = reg.GetConfigKeysByGroup(customGroup)
  assert.Nil(t, err)
  assert.Equal(t, 1, keys.Size())
  assert.True(t, keys.Contains(key))