meshregistry: update go-zk to fix infinite re-connect and optimize re…

staging/src/slime.io/slime/modules/bundle-hango/go.mod

@@ -113,7 +113,7 @@ require (
 
 replace (
  github.com/envoyproxy/go-control-plane => github.com/envoyproxy/go-control-plane v0.11.0
- github.com/go-zookeeper/zk => github.com/slime-io/go-zk v0.0.0-20220815023449-add01187ad4f
+ github.com/go-zookeeper/zk => github.com/slime-io/go-zk v0.0.0-20230210055653-e5ab5bf5cb30
  github.com/prometheus/common => github.com/prometheus/common v0.26.0
 
  istio.io/api => istio.io/api v0.0.0-20211206163441-1a632586cbd4

staging/src/slime.io/slime/modules/bundle-hango/go.sum

@@ -453,10 +453,7 @@ github.com/cncf/udpa/go v0.0.0-20200629203442-efcf912fb354/go.mod h1:WmhPx2Nbnht
 github.com/cncf/udpa/go v0.0.0-20201120205902-5459f2c99403/go.mod h1:WmhPx2Nbnhtbo57+VJT5O0JRkEi1Wbu0z5j0R8u5Hbk=
 github.com/cncf/udpa/go v0.0.0-20210930031921-04548b0d99d4 h1:hzAQntlaYRkVSFEfj9OTWlVV1H155FMD8BTKktLv0QI=
 github.com/cncf/udpa/go v0.0.0-20210930031921-04548b0d99d4/go.mod h1:6pvJx4me5XPnfI9Z40ddWsdw2W/uZgQLFXToKeRcDiI=
-github.com/cncf/xds/go v0.0.0-20210312221358-fbca930ec8ed/go.mod h1:eXthEFrGJvWHgFFCl3hGmgk+/aYT6PnTQLykKQRLhEs=
-github.com/cncf/xds/go v0.0.0-20210805033703-aa0b78936158/go.mod h1:eXthEFrGJvWHgFFCl3hGmgk+/aYT6PnTQLykKQRLhEs=
 github.com/cncf/xds/go v0.0.0-20210922020428-25de7278fc84/go.mod h1:eXthEFrGJvWHgFFCl3hGmgk+/aYT6PnTQLykKQRLhEs=
-github.com/cncf/xds/go v0.0.0-20211001041855-01bcc9b48dfe/go.mod h1:eXthEFrGJvWHgFFCl3hGmgk+/aYT6PnTQLykKQRLhEs=
 github.com/cncf/xds/go v0.0.0-20211011173535-cb28da3451f1/go.mod h1:eXthEFrGJvWHgFFCl3hGmgk+/aYT6PnTQLykKQRLhEs=
 github.com/cncf/xds/go v0.0.0-20220314180256-7f1daf1720fc h1:PYXxkRUBGUMa5xgMVMDl62vEklZvKpVaxQeN9ie7Hfk=
 github.com/cncf/xds/go v0.0.0-20220314180256-7f1daf1720fc/go.mod h1:eXthEFrGJvWHgFFCl3hGmgk+/aYT6PnTQLykKQRLhEs=
@@ -507,15 +504,6 @@ github.com/elazarl/goproxy v0.0.0-20180725130230-947c36da3153/go.mod h1:/Zj4wYkg
 github.com/emicklei/go-restful v0.0.0-20170410110728-ff4f55a20633/go.mod h1:otzb+WCGbkyDHkqmQmT5YD2WR4BBwUdeQoFo8l/7tVs=
 github.com/emicklei/go-restful v2.9.5+incompatible h1:spTtZBk5DYEvbxMVutUuTyh1Ao2r4iyvLdACqsl/Ljk=
 github.com/emicklei/go-restful v2.9.5+incompatible/go.mod h1:otzb+WCGbkyDHkqmQmT5YD2WR4BBwUdeQoFo8l/7tVs=
-github.com/envoyproxy/go-control-plane v0.9.0/go.mod h1:YTl/9mNaCwkRvm6d1a2C3ymFceY/DCBVvsKhRF0iEA4=
-github.com/envoyproxy/go-control-plane v0.9.1-0.20191026205805-5f8ba28d4473/go.mod h1:YTl/9mNaCwkRvm6d1a2C3ymFceY/DCBVvsKhRF0iEA4=
-github.com/envoyproxy/go-control-plane v0.9.4/go.mod h1:6rpuAdCZL397s3pYoYcLgu1mIlRU8Am5FuJP05cCM98=
-github.com/envoyproxy/go-control-plane v0.9.7/go.mod h1:cwu0lG7PUMfa9snN8LXBig5ynNVH9qI8YYLbd1fK2po=
-github.com/envoyproxy/go-control-plane v0.9.9-0.20201210154907-fd9021fe5dad/go.mod h1:cXg6YxExXjJnVBQHBLXeUAgxn2UodCpnH306RInaBQk=
-github.com/envoyproxy/go-control-plane v0.9.9-0.20210217033140-668b12f5399d/go.mod h1:cXg6YxExXjJnVBQHBLXeUAgxn2UodCpnH306RInaBQk=
-github.com/envoyproxy/go-control-plane v0.9.9-0.20210512163311-63b5d3c536b0/go.mod h1:hliV/p42l8fGbc6Y9bQ70uLwIvmJyVE5k4iMKlh8wCQ=
-github.com/envoyproxy/go-control-plane v0.9.10-0.20210907150352-cf90f659a021/go.mod h1:AFq3mo9L8Lqqiid3OhADV3RfLJnjiw63cSpi+fDTRC0=
-github.com/envoyproxy/go-control-plane v0.10.2-0.20220325020618-49ff273808a1/go.mod h1:KJwIaB5Mv44NWtYuAOFCVOjcI94vtpEz2JU/D2v6IjE=
 github.com/envoyproxy/go-control-plane v0.11.0 h1:jtLewhRR2vMRNnq2ZZUoCjUlgut+Y0+sDDWPOfwOi1o=
 github.com/envoyproxy/go-control-plane v0.11.0/go.mod h1:VnHyVMpzcLvCFt9yUz1UnCwHLhwx1WguiVDV7pTG/tI=
 github.com/envoyproxy/protoc-gen-validate v0.1.0/go.mod h1:iSmxcyjqTsJpI2R4NaDN7+kN2VEUnK/pcBlmesArF7c=
@@ -955,7 +943,6 @@ github.com/prometheus/client_golang v1.12.2/go.mod h1:3Z9XVyYiZYEO+YQWt3RD2R3jrb
 github.com/prometheus/client_model v0.0.0-20180712105110-5c3871d89910/go.mod h1:MbSGuTsp3dbXC40dX6PRTWyKYBIrTGTE9sqQNg2J8bo=
 github.com/prometheus/client_model v0.0.0-20190115171406-56726106282f/go.mod h1:MbSGuTsp3dbXC40dX6PRTWyKYBIrTGTE9sqQNg2J8bo=
 github.com/prometheus/client_model v0.0.0-20190129233127-fd36f4220a90/go.mod h1:xMI15A0UPsDsEKsMN9yxemIoYk6Tm2C1GtYGdfGttqA=
-github.com/prometheus/client_model v0.0.0-20190812154241-14fe0d1b01d4/go.mod h1:xMI15A0UPsDsEKsMN9yxemIoYk6Tm2C1GtYGdfGttqA=
 github.com/prometheus/client_model v0.2.0/go.mod h1:xMI15A0UPsDsEKsMN9yxemIoYk6Tm2C1GtYGdfGttqA=
 github.com/prometheus/client_model v0.3.0 h1:UBgGFHqYdG/TPFD1B1ogZywDqEkwp3fBMvqdiQ7Xew4=
 github.com/prometheus/client_model v0.3.0/go.mod h1:LDGWKZIo7rky3hgvBe+caln+Dr3dPggB5dvjtD7w9+w=
@@ -991,8 +978,8 @@ github.com/sirupsen/logrus v1.6.0/go.mod h1:7uNnSEd1DgxDLC74fIahvMZmmYsHGZGEOFrf
 github.com/sirupsen/logrus v1.7.0/go.mod h1:yWOB1SBYBC5VeMP7gHvWumXLIWorT60ONWic61uBYv0=
 github.com/sirupsen/logrus v1.8.1 h1:dJKuHgqk1NNQlqoA6BTlM1Wf9DOH3NBjQyu0h9+AZZE=
 github.com/sirupsen/logrus v1.8.1/go.mod h1:yWOB1SBYBC5VeMP7gHvWumXLIWorT60ONWic61uBYv0=
-github.com/slime-io/go-zk v0.0.0-20220815023449-add01187ad4f h1:VdlfK5ahhz9L3q5zGm7pnRXSWmSUJ4vkDQuP32N7rBE=
-github.com/slime-io/go-zk v0.0.0-20220815023449-add01187ad4f/go.mod h1:nOB03cncLtlp4t+UAkGSV+9beXP/akpekBwL+UX1Qcw=
+github.com/slime-io/go-zk v0.0.0-20230210055653-e5ab5bf5cb30 h1:K+S7aE77oOdwPGm9fOIyohJU961Blf1Wqlyjs1y9s40=
+github.com/slime-io/go-zk v0.0.0-20230210055653-e5ab5bf5cb30/go.mod h1:nOB03cncLtlp4t+UAkGSV+9beXP/akpekBwL+UX1Qcw=
 github.com/slime-io/istio-mcp v0.0.0-20230105060927-109703897996 h1:FfS2cks/HV9zb4VBm0+8qMicX6eRcCBVr+A9I4tApWk=
 github.com/slime-io/istio-mcp v0.0.0-20230105060927-109703897996/go.mod h1:vHYwJsTi/0FUGT8emU8ztlr84sc3G391rBm938dKvjM=
 github.com/slime-io/libistio v0.0.0-20221214030325-22f5add50855 h1:dwzP73IM2BVPeTAr5dZmuu9N1FNj/3K2PxYcEEF8BqE=

staging/src/slime.io/slime/modules/meshregistry/go.mod

@@ -33,7 +33,7 @@ replace (
 
  github.com/envoyproxy/go-control-plane => github.com/envoyproxy/go-control-plane v0.9.5-0.20200326174812-e8bd2869ff56
 
- github.com/go-zookeeper/zk => github.com/slime-io/go-zk v0.0.0-20220815023449-add01187ad4f
+ github.com/go-zookeeper/zk => github.com/slime-io/go-zk v0.0.0-20230210055653-e5ab5bf5cb30
 
  istio.io/api => istio.io/api v0.0.0-20211206163441-1a632586cbd4
  istio.io/istio-mcp => github.com/slime-io/istio-mcp v0.0.0-20230105060927-109703897996

staging/src/slime.io/slime/modules/meshregistry/go.sum

@@ -610,8 +610,8 @@ github.com/sirupsen/logrus v1.6.0/go.mod h1:7uNnSEd1DgxDLC74fIahvMZmmYsHGZGEOFrf
 github.com/sirupsen/logrus v1.7.0/go.mod h1:yWOB1SBYBC5VeMP7gHvWumXLIWorT60ONWic61uBYv0=
 github.com/sirupsen/logrus v1.8.1 h1:dJKuHgqk1NNQlqoA6BTlM1Wf9DOH3NBjQyu0h9+AZZE=
 github.com/sirupsen/logrus v1.8.1/go.mod h1:yWOB1SBYBC5VeMP7gHvWumXLIWorT60ONWic61uBYv0=
-github.com/slime-io/go-zk v0.0.0-20220815023449-add01187ad4f h1:VdlfK5ahhz9L3q5zGm7pnRXSWmSUJ4vkDQuP32N7rBE=
-github.com/slime-io/go-zk v0.0.0-20220815023449-add01187ad4f/go.mod h1:nOB03cncLtlp4t+UAkGSV+9beXP/akpekBwL+UX1Qcw=
+github.com/slime-io/go-zk v0.0.0-20230210055653-e5ab5bf5cb30 h1:K+S7aE77oOdwPGm9fOIyohJU961Blf1Wqlyjs1y9s40=
+github.com/slime-io/go-zk v0.0.0-20230210055653-e5ab5bf5cb30/go.mod h1:nOB03cncLtlp4t+UAkGSV+9beXP/akpekBwL+UX1Qcw=
 github.com/slime-io/istio-mcp v0.0.0-20230105060927-109703897996 h1:FfS2cks/HV9zb4VBm0+8qMicX6eRcCBVr+A9I4tApWk=
 github.com/slime-io/istio-mcp v0.0.0-20230105060927-109703897996/go.mod h1:vHYwJsTi/0FUGT8emU8ztlr84sc3G391rBm938dKvjM=
 github.com/slime-io/libistio v0.0.0-20221214030325-22f5add50855 h1:dwzP73IM2BVPeTAr5dZmuu9N1FNj/3K2PxYcEEF8BqE=

staging/src/slime.io/slime/modules/meshregistry/pkg/source/zookeeper/source.go

@@ -54,7 +54,7 @@ type Source struct {
  zkGatewayModel bool
  patchLabel bool
  ignoreLabels map[string]string
- watchingRoot bool
+ watchingRoot bool // TODO useless?
  watchingWorkerCount int
 
  serviceCache map[string]*ServiceEntryWithMeta
@@ -121,25 +121,38 @@ func NewSource(args bootstrap.ZookeeperSourceArgs, exceptedResources []collectio
  return ret, ret.cacheJson, ret.simpleCacheJson, nil
 }
 
-func (s *Source) reConFunc(event zk.Event) {
- // TODO refactor recon logic
+func (s *Source) reConFunc(reconCh chan<- struct{}) {
+ if s.watchingRoot {
+ return // ??
+ }
 
- switch event.Type {
- case zk.EventDisconnected:
- // rewatch
- if !s.watchingRoot {
- s.Con.Load().(*zk.Conn).Close()
- for {
- con, _, err := zk.Connect(s.addresses, s.timeout)
- if err != nil {
- scope.Infof("re connect zk error %v", err)
- continue
- }
- // replace the connection
- s.Con.Store(con)
- zk.WithEventCallback(s.reConFunc)(con)
- break
+ var curConn *zk.Conn
+ if v := s.Con.Load(); v != nil {
+ curConn = v.(*zk.Conn)
+ }
+ if curConn != nil {
+ curConn.Close()
+ }
+
+ for {
+ con, _, err := zk.Connect(s.addresses, s.timeout, zk.WithEventCallback(func(ev zk.Event) {
+ if ev.Type != zk.EventDisconnected {
+ return
+ }
+
+ // notify recon
+ select {
+ case reconCh <- struct{}{}:
+ default:
  }
+ }))
+ if err != nil {
+ scope.Infof("re connect zk error %v", err)
+ time.Sleep(time.Second)
+ } else {
+ // replace the connection
+ s.Con.Store(con)
+ break
  }
  }
 }
@@ -197,28 +210,31 @@ func (s *Source) Start() {
  }()
  }
 
- go func() {
+ go func() { // do recon
+ reconCh := make(chan struct{}, 1)
+ reconCh <- struct{}{}
+ starter := &sync.Once{}
+
  for {
- con, _, err := zk.Connect(s.addresses, s.timeout)
- if err != nil {
- scope.Errorf("zk conn %s %v met err %v", s.addresses, s.timeout, err)
- time.Sleep(time.Second)
- continue
+ select {
+ case <-s.stop:
+ return
+ case <-reconCh:
+ log.Infof("recv signal, will call reConFunc")
+ s.reConFunc(reconCh)
+ starter.Do(func() {
+ log.Infof("zk connected, will start fetch-data logic")
+ // s.initPush()
+ // TODO 服务自省模式
+ // go s.doWatchAppication(s.ApplicationRegisterRootNode)
+ if s.isPollingMode() {
+ go s.Polling()
+ } else {
+ go s.Watching()
+ }
+ })
  }
- s.Con.Store(con)
- zk.WithEventCallback(s.reConFunc)(con)
- break
- }
-
- // s.initPush()
- // TODO 服务自省模式
- // go s.doWatchAppication(s.ApplicationRegisterRootNode)
- if s.isPollingMode() {
- go s.Polling()
- } else {
- go s.Watching()
  }
- <-s.stop
  }()
 
  if s.args.EnableDubboSidecar {